序文：LLM から Agent へ — 一つの完全なエンジニアリングパス
大規模言語モデル（Large Language Model, LLM）は、ソフトウェアの構築方法を根本から変えた。しかし、単に API を呼び出してテキスト応答を得るだけでは、まだ第一歩に過ぎない。真のエンジニアリング上の課題は、「会話できる」状態から「実行できる」状態へとモデルを進化させること — すなわち、外部ツールの呼び出し、自律的なタスク計画、チーム内での協調、ユーザー嗜好の記憶、そして最終的に本番環境での安定稼働を実現することである。
このパスは八つの中核領域をカバーする：
- LLM 基礎：Tokenization、Transformer アーキテクチャ、コンテキストウィンドウとプロンプトエンジニアリングの理解
- RAG の原理と実践：検索拡張生成を習得し、モデルにプライベート知識へのアクセスを提供する
- Agent ツール呼び出し：Function Calling、ReAct ループ、MCP プロトコル
- Agent 計画と実行：リフレクション機構、Plan & Execute、ワークフローオーケストレーション
- マルチ Agent 協調：階層型協調、黒板モデル、協調方法論
- Memory と Skill：短期・長期記憶管理、Skill システム設計
- Agent 評価：エンドツーエンド評価、ホワイトボックス評価、評価駆動反復
- 本番運用：モデルデプロイメント、推論最適化、安全ガードレール、Harness Engineering
本ガイドは、Agent エンジニアリングを体系的に習得したい開発者向けに、基盤原理から本番実践までの全景的な技術リファレンスを提供する。
Tokenization の原理
なぜ Tokenization が必要か
コンピュータは人間の文字を直接理解できない。大規模モデルがテキストを処理する第一歩は、自然言語を機械が演算可能な数値形式に変換することである。このプロセスを Tokenization（トークン化） と呼ぶ。
Token とは、Tokenizer（トークナイザ）がテキストをエンコードして得られる基本単位であり、各 token は語彙表中の一つの整数 ID に対応する。重要な認識：token は通常、サブワード断片または文字断片であり、必ずしも完全な「単語」とは限らず、また独立した意味を持つとも限らない。
入力テキスト: "ACP is a very"
↓ Tokenizer
Token系列: [347, 1186, 374, 1134]英語を例にとると、単語は語幹と接尾辞に分解されることがある。中国語は文字またはよく使われる単語グループ単位で分割されることがある。空白や句読点も token にエンコードされることがある。モデルによって Tokenizer は大きく異なる — GPT シリーズは BPE（Byte Pair Encoding）を使用し、LLaMA は SentencePiece BPE を使用し、一部の中国語モデルは中国語コーパス向けに特別な最適化を施している。
Token ベクトル化と位置エンコーディング
整数 ID そのものの数値的大きさには意味的な意義はない。ID 500 が ID 50 より「重要」というわけではない。したがって、これらの離散的な ID を Embedding 行列 を通じて密なベクトルに写像する必要がある。
Token ID → Embedding Matrix Lookup → d-dimensional vector
1186 → [0.023, -0.451, 0.789, ..., -0.312] (d=4096 あるいはそれ以上)同時に、言語の順序は極めて重要である — 「私があなたを助ける」と「あなたが私を助ける」は全く異なる。Transformer の Self-Attention 機構自体は系列の順序を感知する能力を持たないため、追加で 位置エンコーディング（Positional Encoding） を付与する必要がある。現代のモデルは通常、回転位置エンコーディング（RoPE, Rotary Position Embedding）を使用し、Attention 計算に相対位置情報を導入することで、モデルが自然に長系列を処理できるようにする。
デコード戦略：確率から出力へ
モデル推論後、logits（スコアベクトル）が得られ、softmax を経て確率分布 P(next_token | context) に変換される。その後、デコード戦略によって出力 token を選択する：
| 戦略 | 原理 | 適用シーン |
|---|---|---|
| 貪欲デコード（Greedy） | 毎回最も確率の高い token を選択 | 決定論的な出力が必要なタスク（コード生成、構造化抽出など） |
| Beam Search | 複数の候補パスを保持し、全体確率が最も高い系列を選択 | 翻訳、要約など大域的最適が必要なタスク |
| Top-k Sampling | 確率上位 k 個の token からランダムサンプリング | クリエイティブライティング、対話生成 |
| Top-p（Nucleus） | 累積確率が p を超える最小の token 集合からサンプリング | 汎用対話、多様性と品質のバランス |
出力のランダム性を制御する二つの中核パラメータ：
Temperature（温度）：softmax 確率分布の「尖り具合」を制御する。
# temperature の効果比較
# 元の logits → softmax(logits / temperature)
# T=0.1: 確率が高度に集中、ほぼ決定論的出力
# T=0.7: 適度、合理的な多様性を保持
# T=1.5: 確率が均等に近づき、出力は高度にランダムTop_p（核サンプリング閾値）：サンプリングに参加する候補 token の範囲を制御する。例えば top_p=0.9 は、累積確率が 90% に達する最小の token 集合からのみサンプリングすることを意味する。
# 典型的な設定例
response = client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": "人工知能についての詩を書いてください"}],
temperature=0.8, # 適度な創造性
top_p=0.9, # 核サンプリング
max_tokens=200
)自己回帰生成と停止条件
モデルは**自己回帰生成（Autoregressive Generation）**を採用する：毎回新しい token を一つ生成し、それを入力末尾に追加し、新しい系列に基づいて次の token を予測し続ける。
"ACP is a very" → "informative" を予測
"ACP is a very informative" → "course" を予測
"ACP is a very informative course" → ... → <EOS> を予測 → 停止停止条件は以下を含む：
- 特殊な**終端記号（EOS token）**の生成
- 事前設定された
max_tokens制限への到達 - ユーザー指定の停止語系列の生成
モデルシリーズによって終端記号は異なる：
| モデル | 終端記号 |
|---|---|
| GPT シリーズ | <|endoftext|> |
| LLaMA/Mistral | </s> |
| DeepSeek | <｜end▁of▁sentence｜> |
| 一部の中国語大規模モデル | <|im_end|> |
ストリーミング出力（Streaming）の本質は、サーバー側が一つまたは数個の token を生成するたびに即座にデコードして増分送信することであり、すべての token 生成が完了してから返すわけではない。
Transformer アーキテクチャと Attention 機構
アーキテクチャ全景
現代の大規模言語モデルの中核は Transformer アーキテクチャである。完全な Transformer はエンコーダ（Encoder）とデコーダ（Decoder）を含むが、現在主流の自己回帰 LLM（GPT、LLaMA など）は**デコーダのみ（Decoder-only）**のアーキテクチャを使用する。
因果自己注意（Causal Self-Attention）
Attention 機構の中核公式：
Attention(Q, K, V) = softmax(QK^T / √d_k) · Vここで Q（Query）、K（Key）、V（Value）はいずれも同一入力系列から線形変換によって得られる。主要な設計：
（1）スケーリング因子 √d_k：ドット積の結果が大きくなりすぎて softmax の勾配消失を防ぐ。ベクトル次元 d_k が大きい場合、ドット積値の分散も増大するため、√d_k で除算して分散を正規化する。
（2）因果マスク（Causal Mask）：自己回帰モデルでは、各 token は自分より前の token だけを「見る」ことができ、未来の内容を「覗き見る」ことはできない。これは上三角行列の位置に -∞ を埋めることで実現される：
入力: ["ACP", "is", "a", "very"]
Attention 行列 (因果マスク適用後):
ACP is a very
ACP 0.8 -∞ -∞ -∞
is 0.3 0.7 -∞ -∞
a 0.2 0.3 0.5 -∞
very 0.1 0.2 0.3 0.4（3）マルチヘッド注意（Multi-Head Attention）：一度だけ Attention 計算を行うのではなく、複数組の Q/K/V 射影を並列実行し、各組が異なる意味関係（文法構造、指示関係、意味的類似性など）に注目し、最後に全ヘッドの出力を連結する。
# マルチヘッド注意の疑似コード
def multi_head_attention(x, num_heads=8, d_model=512):
d_head = d_model // num_heads # 各ヘッドの次元
outputs = []
for h in range(num_heads):
Q = linear_projection(x, d_head)
K = linear_projection(x, d_head)
V = linear_projection(x, d_head)
attn_out = softmax(Q @ K.T / sqrt(d_head)) @ V
outputs.append(attn_out)
return concat(outputs) # 全ヘッドを連結フィードフォワードネットワーク（Feed-Forward Network）
各 Attention 層の後には FFN が続く：
FFN(x) = GELU(x·W₁ + b₁) · W₂ + b₂FFN は通常、隠れ次元をまず拡大（例：4x）し、その後元の次元に圧縮する。この「拡張-圧縮」構造がモデルに非線形変換能力を提供し、モデルが知識を蓄積・運用する上で重要なコンポーネントである。
残差接続と層正規化
各サブレイヤー（Attention と FFN）は残差接続によって入力と加算される：
output = LayerNorm(x + Sublayer(x))残差接続により勾配が浅い層まで直接伝播し、深層ネットワークの学習困難を解決する。現代アーキテクチャの層正規化（LayerNorm）は通常、Pre-Norm 配置（サブレイヤーの前で正規化）を採用し、元の Post-Norm よりも安定している。
コンテキストウィンドウと Token 予算
コンテキストウィンドウの本質
大規模モデルが入力を受け取る場所を**コンテキストウィンドウ（Context Window）**と呼ぶ。これはコンピュータのメモリ（RAM）に例えられる — 容量は有限であり、パフォーマンスに直接影響する。
現代モデルのコンテキストウィンドウは大幅に拡張されている：
- GPT-4 Turbo: 128K tokens
- Claude 3: 200K tokens
- Gemini 1.5 Pro: 1M+ tokens
- オープンソースモデル（LLaMA 3、一部の中国語モデル等）: 32K–128K tokens
しかしウィンドウが大きいからといって濫用できるわけではない。研究によれば “Lost in the Middle” 効果が存在する：モデルはコンテキスト中間部分の情報処理能力が著しく低下し、冒頭（初頭効果）と末尾（新近効果）の内容により多くの注意を払う。
Token 予算管理
本番環境では、メモリ管理と同様にコンテキストを管理する必要がある。以下が中核戦略である：
（1）Token 消費の正確な計算
import tiktoken
def count_tokens(text: str, model: str = "gpt-4") -> int:
encoding = tiktoken.encoding_for_model(model)
return len(encoding.encode(text))
# 例：messages の総 token 数を計算
def count_message_tokens(messages):
encoding = tiktoken.encoding_for_model("gpt-4")
total = 0
for msg in messages:
# 各メッセージには固定オーバーヘッドがある（約 4 tokens）
total += 4
total += len(encoding.encode(msg["content"]))
total += 2 # 応答のプライミング
return total（2）コンテキストウィンドウの割り当て戦略
┌──────────────────────────────────────────────┐
│ Token 予算割り当ての推奨（128K ウィンドウの場合） │
├──────────────────────────────────────────────┤
│ System Prompt: 2-5K (ロール定義、ルール) │
│ RAG 検索結果: 3-8K (関連知識断片) │
│ 対話履歴: 10-20K (直近 N ターン) │
│ 現在のユーザー入力: 1-3K │
│ 応答用予約領域: 4-8K │
│ バッファ余裕: 残り (~80K) │
└──────────────────────────────────────────────┘（3）コンテキストエンジニアリング（Context Engineering）
コンテキストエンジニアリングとは、コンテキストを体系的に設計・構築・最適化する実践である。単に「情報を prompt に詰め込む」のではなく、四つの中核技術を含む：
| 技術 | 解決する問題 | 中核手法 |
|---|---|---|
| RAG | プライベート知識の不足 | 外部知識ベースから関連情報を検索してコンテキストに注入 |
| Prompt Engineering | 指示の精度不足 | 入念に設計された指示によってモデルの振る舞いを誘導 |
| Tool Use | モデルが操作を実行できない | モデルに外部ツールを呼び出す能力を付与 |
| Memory | セッション間の忘却 | 長期・短期記憶機構の確立 |
多くの大規模モデルアプリケーションの失敗は、モデル自体の知能不足ではなく、「コンテキスト」の失敗である。コンテキストエンジニアリングこそが、大規模モデルの潜在力を解放する鍵である。
プロンプトエンジニアリング方法論
System Prompt 設計
System Prompt はモデルの「憲法」である — ロールの行動境界、回答スタイル、タスク制約を定義する。優れた System Prompt は以下を含むべきである：
# System Prompt 構造テンプレート
ロール定義: |
あなたは経験豊富な Python 技術ドキュメント審査員であり、
コードの正確性と教育的有効性に注力しています。
行動準則:
- コード内の変数名、API バージョン番号を変更しないこと
- 問題を発見した場合は具体的な位置と修正提案を提示すること
- 判断に情報が不足している場合は、明確に「不確か」と述べること
出力形式:
## 審査レポート
### 重要問題
- **[第 N 行]**: 問題の説明
- 重大度: 重大|一般|軽微
- 修正提案: 具体的な提案
制約条件:
- 冗談や不要な情景描写を禁止する
- 用語の初出時には説明を付けること
- コードブロックには必要な import 文を含めることFew-Shot と構造化出力
Few-Shot 例示：入力-出力の範例を提供し、モデルに形式とスタイルを模倣させる。
examples = [
{
"input": "Python デコレータとは何か説明してください",
"output": "### 痛点の導入\n複数の関数に同じログ記録ロジックを追加したいと思ったことはありませんか？..."
},
{
"input": "リスト内包表記とは何か説明してください",
"output": "### 痛点の導入\nリストから偶数だけをフィルタリングするために5行の for ループを書いたことはありませんか？..."
}
]
prompt = f"""
以下の例のスタイルに従ってユーザーの質問に回答してください：
{examples}
ユーザーの質問: {user_question}
"""構造化出力：JSON Schema または Pydantic モデルによって出力形式を制約する。
from pydantic import BaseModel
from typing import List, Optional
class CodeReview(BaseModel):
file_name: str
issues: List[dict]
overall_score: int # 1-5
requires_rewrite: bool
# prompt に Schema を添付
prompt = f"""
以下の JSON Schema に従って審査結果を出力してください：
{CodeReview.model_json_schema()}
審査対象コード:
{code}
"""Chain-of-Thought（思考連鎖）
複数ステップの推論が必要な複雑なタスクでは、モデルに「思考プロセスを声に出す」よう誘導することで精度が大幅に向上する。
# ❌ 直接要求（精度が低い）
prompt_simple = "計算：クラスに30人の生徒がいて、男子は女子より4人多い。男子は何人か？"
# ✅ CoT プロンプト（精度が高い）
prompt_cot = """
計算：クラスに30人の生徒がいて、男子は女子より4人多い。男子は何人か？
段階的に推論してください：
ステップ 1: 女子の人数を x とすると、男子の人数は x + 4
ステップ 2: 総人数は x + (x + 4) = 30
ステップ 3: 方程式 2x + 4 = 30 を解くと、x = 13
ステップ 4: 男子の人数 = x + 4 = 17
答え: 17 人
"""CoT の変種には以下も含まれる：
- ToT（Tree of Thoughts）：複数の推論パスを同時に探索し、最適なものを選択
- GoT（Graph of Thoughts）：推論を有向グラフとして表現し、より複雑な推論トポロジーをサポート
- Self-Consistency：CoT パスを複数回サンプリングし、多数決の結果を採用
Meta Prompting：モデルに自身のプロンプトを最適化させる
一度で完璧なプロンプトを書くことはほぼ不可能である。Meta Prompting の中核思想は：大規模モデルに「プロンプト審査専門家」を演じさせ、プロンプト自体の分析と最適化を支援させることである。
meta_prompt = """
あなたはプロンプトエンジニアリングの専門家です。以下のプロンプトの欠陥を分析し、最適化バージョンを生成してください。
現在のプロンプト:
{current_prompt}
このプロンプトの出力:
{current_output}
期待する出力:
{desired_output}
ギャップを分析し、最適化されたプロンプトを出力してください。
"""
# このループは自動化できる：生成 → 評価 → 最適化 → 再生成完全な Meta Prompting フローでは「参考答案」と定量スコアリングも導入できる：
- 参考答案の設定：理想的な出力を定義する
- ギャップ分析：「評価者」モデルに生成結果と参考答案を比較させる
- プロンプト最適化：ギャップ分析レポートに基づいてプロンプトを書き直す
- 定量検証：採点者（Grader）を用いて複数バージョンを採点する
Embedding とベクトル検索
Embedding モデルの動作原理
Embedding モデルはテキストを高次元ベクトルに変換し、意味的に類似したテキストがベクトル空間内で近接するようにする。
"私はリンゴを食べるのが好き" → [0.12, -0.34, 0.56, ..., 0.78] (1024次元)
"私はリンゴが大好き" → [0.11, -0.33, 0.55, ..., 0.79] ← 距離が非常に近い
"自動車修理ガイド" → [-0.78, 0.45, -0.23, ..., 0.01] ← 距離が非常に遠いEmbedding モデルの学習は通常、対照学習（Contrastive Learning） の段階を含む：入力は関連/非関連とラベル付けされた多数のテキストペアであり、学習目標は関連テキストのベクトル類似度を最大化し、非関連テキストの類似度を最小化することである。
# 二つのテキストベクトルのコサイン類似度を計算
import numpy as np
def cosine_similarity(a, b):
return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))
# 例
query_vec = embedding_model.encode("年次休暇の申請方法は？")
doc_vec = embedding_model.encode("従業員年次休暇申請フロー")
similarity = cosine_similarity(query_vec, doc_vec)
print(f"類似度: {similarity:.4f}") # 0.92 — 高度に関連ベクトルデータベースの選定
ベクトルデータベースは RAG システムの中核インフラストラクチャである。選択時には以下のトレードオフが必要である：
| 方式 | 代表製品 | 利点 | 欠点 | 適用シーン |
|---|---|---|---|---|
| インメモリ | LlamaIndex 内蔵 | ゼロ設定、迅速なプロトタイピング | データが永続化されず、メモリ制限を受ける | 開発テスト |
| ローカルベクトルDB | Milvus, Qdrant, Chroma | 機能が完全、データ制御可能 | 自身でデプロイ・保守が必要 | 中小規模アプリケーション |
| マネージドサービス | Pinecone, Weaviate Cloud | 運用不要、自動スケーリング | コストが高く、データが外部にある | 本番環境、弾力的需要 |
| 既存DB拡張 | PostgreSQL + pgvector, Elasticsearch | 既存インフラを活用 | ベクトル性能は専用DBに劣る | 該当DBを既に保有するチーム |
# Chroma 使用例（軽量ローカルベクトルDB）
import chromadb
from chromadb.utils import embedding_functions
client = chromadb.PersistentClient(path="./chroma_db")
collection = client.create_collection(
name="company_docs",
embedding_function=embedding_functions.OpenAIEmbeddingFunction(
api_key="your-api-key",
model_name="text-embedding-3-small"
)
)
# ドキュメント追加
collection.add(
documents=["従業員年次休暇申請フロー...", "出張精算基準..."],
metadatas=[{"source": "hr_policy.pdf"}, {"source": "finance_policy.pdf"}],
ids=["doc_1", "doc_2"]
)
# 検索
results = collection.query(
query_texts=["年次休暇の申請方法は？"],
n_results=3
)ドキュメント分割戦略
分割の基本矛盾
RAG システムの検索効果はドキュメント分割の品質に大きく依存する。分割の中核的矛盾は：
チャンクが大きすぎる → 検索時に過剰なノイズを導入し、モデルの注意が希釈される
チャンクが小さすぎる → 重要な情報が切断され、コンテキストを失う可能性があるあらゆる場面に通用する「万能の」最適分割戦略は存在しない。ドキュメントタイプ、検索シーン、モデル能力に応じて選択する必要がある。
五つの主流分割手法
Token 分割
固定 token 数で分割し、token 消費を正確に制御する必要があるシーンに適する。
from llama_index.core.node_parser import TokenTextSplitter
splitter = TokenTextSplitter(
chunk_size=256, # 各チャンクの token 数
chunk_overlap=30 # 隣接チャンクの重複 token 数
)
nodes = splitter.get_nodes_from_documents(documents)利点：コンテキストサイズを正確に制御でき、コンテキストウィンドウが小さいモデルに適する。 欠点：文の途中で切断され、意味的完全性を損なう可能性がある。
文分割
文の完全性を保持する分割方式であり、ほとんどのシーンでデフォルトの選択肢となる。
from llama_index.core.node_parser import SentenceSplitter
splitter = SentenceSplitter(
chunk_size=512,
chunk_overlap=50
)利点：自然言語の意味単位の完全性を保持する。 欠点：ドキュメント構造を感知せず、関連する段落を異なるチャンクに分割する可能性がある。
文ウィンドウ分割
インデックス作成時と検索時で異なる粒度を使用する：インデックスには小粒度で正確なマッチングを、検索返却時には隣接するコンテキストウィンドウを付加する。
from llama_index.core.node_parser import SentenceWindowNodeParser
parser = SentenceWindowNodeParser(
window_size=3, # 検索時に隣接3文を拡張
window_metadata_key="window",
original_text_metadata_key="original"
)中核的利点：検索精度とコンテキスト完全性を両立する。
意味分割
意味的関連性に基づいて適応的に分割点を選択し、ドキュメントの意味的連続性を保持する。
from llama_index.core.node_parser import SemanticSplitterNodeParser
splitter = SemanticSplitterNodeParser(
buffer_size=1,
breakpoint_percentile_threshold=95, # 類似度がこの閾値を下回ったら分割
embed_model=embed_model
)適用シーン：論理性が高く、内容が専門的な長文ドキュメント。
Markdown 分割
Markdown 構造化ドキュメントに特化して最適化され、見出し階層に従って分割する。
from llama_index.core.node_parser import MarkdownNodeParser
parser = MarkdownNodeParser()
# #、##、### などの見出し階層を自動認識し、各見出し段落で分割ベストプラクティス：ドキュメントを PDF/Word から Markdown に変換してから分割し、見出し構造を活用して検索精度を向上させる。
分割戦略選択ガイド
| ドキュメントタイプ | 推奨戦略 | 理由 |
|---|---|---|
| 技術マニュアル（構造が明確） | Markdown 分割 | 見出し階層を活用して構造を保持 |
| 法律契約書（論理が厳密） | 意味分割 | 条項の意味的完全性を保持 |
| 対話記録 | 文ウィンドウ分割 | 前後の文脈による意味理解が必要 |
| コードドキュメント | Token 分割 + 意味分割 | 長さの正確な制御が必要 |
| ニュース/ブログ | 文分割 | 段落間の関連が弱い |
検索拡張生成パイプライン
RAG の二段階アーキテクチャ
RAG（Retrieval-Augmented Generation）は、大規模モデルの「知識不足」を解決する中核アーキテクチャである。プロセスを二段階に分ける：
段階一：インデックス構築
- ドキュメント解析：PDF、Word、Markdown などの形式をプレーンテキストに解析
- テキスト分割：選択した戦略に従ってドキュメントを段落に分割
- ベクトル化：Embedding モデルを用いて各チャンクをベクトルに変換
- インデックス保存：ベクトルをベクトルデータベースに保存し、インデックスを構築
段階二：検索と生成
┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐
│ ユーザー質問 │ → │ ベクトル検索 │ → │ Prompt組立 │ → │ モデル生成 │
│ Query │ │ Retrieve │ │ Augment │ │ Generate │
└──────────┘ └──────────┘ └──────────┘ └──────────┘- ユーザー質問：ユーザーの質問を受け取る
- ベクトル検索：質問をベクトル化し、ベクトルデータベース内で最も類似したチャンクを検索
- Prompt 組立：検索された知識断片 + 元の質問 + 指示を完全な Prompt に組み立て
- モデル生成：大規模モデルが強化されたコンテキストに基づいて回答を生成
完全な RAG パイプライン例
from openai import OpenAI
import numpy as np
client = OpenAI()
class SimpleRAG:
def __init__(self, embed_model="text-embedding-3-small"):
self.embed_model = embed_model
self.documents = [] # ドキュメントテキストを保存
self.embeddings = [] # ドキュメントベクトルを保存
def add_documents(self, docs: list[str]):
"""インデックス構築：ベクトル化してドキュメントを保存"""
for doc in docs:
vec = self._embed(doc)
self.documents.append(doc)
self.embeddings.append(vec)
def _embed(self, text: str) -> np.ndarray:
resp = client.embeddings.create(
model=self.embed_model,
input=text
)
return np.array(resp.data[0].embedding)
def retrieve(self, query: str, top_k: int = 3) -> list[str]:
"""最も関連するドキュメント断片を検索"""
query_vec = self._embed(query)
similarities = [
np.dot(query_vec, doc_vec) /
(np.linalg.norm(query_vec) * np.linalg.norm(doc_vec))
for doc_vec in self.embeddings
]
top_indices = np.argsort(similarities)[-top_k:][::-1]
return [self.documents[i] for i in top_indices]
def query(self, question: str) -> str:
"""完全な RAG クエリ"""
contexts = self.retrieve(question)
prompt = f"""以下の参考情報に基づいて質問に回答してください：
参考情報：
{' '.join(contexts)}
質問：{question}
参考情報が質問に答えるのに不十分な場合は、その旨を明示してください。"""
resp = client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": prompt}]
)
return resp.choices[0].message.contentRAG 複数ターン対話における質問書き換え
RAG シーンで複数ターン対話を実現する際には独自の課題がある。ユーザーが第二ターンで「彼の上司は誰ですか？」と言った場合、この文をそのまま検索に使うと完全に失敗する — システムは「彼」が誰を指すかわからない。
解決策：質問書き換え（Query Rewriting）
def rewrite_query(conversation_history: list, current_query: str) -> str:
"""大規模モデルを使用してコンテキスト依存の質問を独立した質問に書き換え"""
rewrite_prompt = f"""
対話履歴に基づいて、現在の質問をコンテキストに依存しない独立した質問に書き換えてください。
対話履歴:
{format_history(conversation_history)}
現在の質問: {current_query}
書き換え後の質問:"""
resp = client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": rewrite_prompt}],
temperature=0.1
)
return resp.choices[0].message.content
# 例
# 履歴: ユーザー「張三の席はどこですか？」 アシスタント「A棟5階です」
# 現在: 「彼の上司は誰ですか？」
# 書き換え後: 「張三の上司は誰ですか？」高度な RAG パターン
HyDE（Hypothetical Document Embeddings）
HyDE の中核思想：まずモデルに仮想的な回答を「でっち上げ」させ、その仮想回答を用いて検索を行う（元の質問ではなく）。この直感は — 仮想回答が元の質問よりも意味的に実際のドキュメントに近い、という点にある。
def hyde_retrieve(query: str, top_k: int = 3) -> list[str]:
"""HyDE 手法を用いた検索"""
# Step 1: 仮想回答を生成
hyde_prompt = f"""
Question: {query}
Please write a passage that answers this question.
Passage:"""
hyde_resp = client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": hyde_prompt}]
)
hypothetical_doc = hyde_resp.choices[0].message.content
# Step 2: 元の質問ではなく仮想回答で検索
query_vec = embed(hypothetical_doc)
results = vector_db.search(query_vec, top_k=top_k)
return results
中核原則：
- アルゴリズムを最適化する前に、まず不足している知識を補充する
- 再現率を改善する前に、まずドキュメント品質を向上させる
- ユーザーの意図を継続的に収集し、「データ収集-知識更新-専門家検証」の閉ループを形成する
Function Calling プロトコル
AutoGen Studio — Microsoft のノーコード マルチエージェントワークフロービルダー — microsoft/autogen より
DSPy — LLM を(プロンプトではなく)宣言的にプログラム;Hermes の自己進化パイプラインが使用 — stanfordnlp/dspy より
Function Calling とは
Function Calling（関数呼び出し、Tool Calling とも呼ばれる）は、大規模モデル API が提供する標準機能である。モデルが必要に応じて純粋なテキスト応答ではなく、構造化されたツール呼び出し命令を出力することを可能にする。
ワークフローは以下の通り：
JSON Schema ツール定義
# ツールリストを定義
tools = [
{
"type": "function",
"function": {
"name": "search_knowledge_base",
"description": "社内ナレッジベースを検索し、ポリシードキュメントと操作ガイドを取得",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "検索キーワードまたは質問"
},
"category": {
"type": "string",
"enum": ["hr", "it", "finance", "general"],
"description": "知識カテゴリ"
}
},
"required": ["query"]
}
}
},
{
"type": "function",
"function": {
"name": "send_email",
"description": "メールを送信",
"parameters": {
"type": "object",
"properties": {
"to": {
"type": "string",
"description": "受信者メールアドレス"
},
"subject": {
"type": "string",
"description": "メール件名"
},
"body": {
"type": "string",
"description": "メール本文"
}
},
"required": ["to", "subject", "body"]
}
}
}
]完全な Function Calling ループ
from openai import OpenAI
import json
client = OpenAI()
def execute_function_call(tool_call):
"""ツール呼び出しを実行し結果を返す"""
func_name = tool_call.function.name
args = json.loads(tool_call.function.arguments)
if func_name == "search_knowledge_base":
# 実際の検索ロジック
result = knowledge_base.search(args["query"])
return json.dumps(result)
elif func_name == "send_email":
# 実際のメール送信ロジック
result = email_service.send(
to=args["to"],
subject=args["subject"],
body=args["body"]
)
return json.dumps({"status": "sent" if result else "failed"})
else:
return json.dumps({"error": f"Unknown function: {func_name}"})
def chat_with_tools(user_message: str, messages: list = None):
"""Function Calling をサポートする対話"""
if messages is None:
messages = [
{"role": "system", "content": "あなたは企業アシスタントです。ナレッジベースの検索とメール送信ができます。"}
]
messages.append({"role": "user", "content": user_message})
# 最初の呼び出し：モデルがツール使用の要否を判断
response = client.chat.completions.create(
model="gpt-4",
messages=messages,
tools=tools
)
assistant_msg = response.choices[0].message
# モデルがツールを呼び出そうとする場合
if assistant_msg.tool_calls:
messages.append(assistant_msg)
for tool_call in assistant_msg.tool_calls:
# ツールを実行
result = execute_function_call(tool_call)
# 結果をモデルに返す
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": result
})
# 二回目の呼び出し：モデルがツール結果に基づいて最終応答を生成
final_response = client.chat.completions.create(
model="gpt-4",
messages=messages
)
return final_response.choices[0].message.content
# モデルが直接応答する場合
return assistant_msg.contentツール定義のベストプラクティス
- 説明は正確に：モデルは
descriptionに基づいてツール呼び出しのタイミングを判断する。説明が曖昧だと誤呼び出しを招く - パラメータには制約を：
enum、required、型制約を用いてモデルのパラメータエラーを減らす - 関数の責務は単一に：一つの関数に複数種類の操作を詰め込まない
- 構造化された結果を返す：ツールの戻り値はモデルが理解しやすいようにし、JSON の使用を推奨
# ❌ 悪いツール説明
{"name": "do_stuff", "description": "操作を実行", "parameters": {...}}
# ✅ 正確なツール説明
{
"name": "cancel_meeting",
"description": "指定された会議をキャンセルする。会議 ID とキャンセル理由が必要",
"parameters": {
"properties": {
"meeting_id": {"type": "string", "description": "会議の一意識別子"},
"reason": {"type": "string", "description": "キャンセル理由、全参加者に通知されます"}
},
"required": ["meeting_id"]
}
}ReAct 推論-行動ループ
ReAct の中核思想
ReAct（Reasoning + Acting）は、モデルに**推論（Thought）と行動（Action）**を交互に行わせるパターンである。一度に最終回答を生成するのではなく、「思考→行動→観察→思考…」のループを通じて問題を解決する。
典型的な ReAct 実行プロセス：
ユーザー: "張三の部署を調べて、彼の上司にメールを送ってください"
Thought 1: まず張三の部署情報を調べる必要がある
Action 1: search_knowledge_base(query="張三 部署")
Observation 1: "張三は教研部に所属、上司は李四([email protected])"
Thought 2: 情報を取得した。次に李四にメールを書く必要がある
Action 2: send_email(to="[email protected]", subject="張三について",
body="...")
Observation 2: {"status": "sent"}
Thought 3: タスク完了
Final Answer: "張三は教研部に所属していることを確認し、彼の上司である李四にメールを送信しました。"手動での ReAct Agent 実装
class ReActAgent:
def __init__(self, tools: dict, max_iterations: int = 10):
self.tools = tools
self.max_iterations = max_iterations
def run(self, task: str) -> str:
messages = [
{"role": "system", "content": self._build_system_prompt()},
{"role": "user", "content": task}
]
for i in range(self.max_iterations):
response = client.chat.completions.create(
model="gpt-4",
messages=messages,
tools=self._format_tools()
)
msg = response.choices[0].message
if msg.content and not msg.tool_calls:
# モデルが最終回答を返した
return msg.content
if msg.tool_calls:
# アシスタントのツール呼び出しを履歴に追加
messages.append(msg)
for tc in msg.tool_calls:
tool_name = tc.function.name
args = json.loads(tc.function.arguments)
# ツールを実行
result = self.tools[tool_name](**args)
# 観察結果を履歴に追加
messages.append({
"role": "tool",
"tool_call_id": tc.id,
"content": json.dumps(result)
})
print(f" [Tool: {tool_name}({args}) → {result}]")
return "ReAct ループが最大反復回数に達しました"
def _build_system_prompt(self) -> str:
return """あなたはツールを使用できる知的アシスタントです。
ReAct パターンに従ってください：まず思考し、次に行動し、結果を観察し、次のステップを決定します。
タスクが完了した場合は、直接最終回答を提示してください。"""
def _format_tools(self) -> list:
return [
{
"type": "function",
"function": {
"name": name,
"description": func.__doc__ or "",
"parameters": get_schema(func)
}
}
for name, func in self.tools.items()
]ReAct の利点と限界
利点：
- 可観測：各ステップの思考と行動の軌跡を追跡可能
- 訂正可能：誤った結果を観察した後に戦略を調整可能
- 組合せ可能：複数のツールを自動的に組み合わせて解決策を構成
限界：
- ループ回数が予測不可能（無限ループに陥る可能性がある）
- 複数回の API 呼び出しにより遅延とコストが増加
- ツールが返す観察結果の品質に依存
MCP プロトコルとツールエコシステム
なぜ MCP が必要か
Function Calling には根本的な問題がある：ツール定義と消費の結合。各 Agent 開発者は自分のコード内でツールの JSON Schema をハードコードする必要がある。ツール API がアップグレードされると、そのツールを統合しているすべての Agent が手動更新を必要とする。
従来の Function Calling パターン:
Agent A ──ハードコードSchema──→ web_search v1
Agent B ──ハードコードSchema──→ web_search v1 ← 重複定義
Agent C ──ハードコードSchema──→ web_search v1 ← 重複定義
MCP パターン:
Agent A ──┐
Agent B ──┼── MCP Client ──→ MCP Server (web_search)
Agent C ──┘ ↑
ツール提供者が Schema を定義MCP（Model Context Protocol） の中核思想は「ツールを提供する者が、ツールを定義する」である。ツール定義の責務を Agent（消費側）からツールサービス（提供側）に移す。
MCP のアーキテクチャ上の役割
| 役割 | 責務 | 類推 |
|---|---|---|
| MCP Server | ツールを宣言（名称、説明、パラメータ）、ツールロジックを実行 | USB デバイス |
| MCP Client | MCP Server に接続、ツール定義を取得、呼び出し要求を送信 | USB ホストコントローラ |
| Agent | MCP Client を使用してツールリストを取得、呼び出しを決定 | アプリケーション |
MCP Server と Client の構築
MCP Server 例：
from mcp.server import Server, stdio_server
from mcp.types import Tool, TextContent
app = Server("web-search")
@app.list_tools()
async def list_tools() -> list[Tool]:
return [
Tool(
name="web_search",
description="インターネットを検索して最新情報を取得",
inputSchema={
"type": "object",
"properties": {
"query": {"type": "string", "description": "検索キーワード"},
"num_results": {"type": "integer", "default": 5}
},
"required": ["query"]
}
)
]
@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
if name == "web_search":
results = search_engine.search(
arguments["query"],
num=arguments.get("num_results", 5)
)
return [TextContent(type="text", text=json.dumps(results))]
raise ValueError(f"Unknown tool: {name}")
# stdio 経由で Server を起動
async def main():
async with stdio_server() as streams:
await app.run(streams[0], streams[1], app.create_initialization_options())
if __name__ == "__main__":
import asyncio
asyncio.run(main())MCP Client 統合例：
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
async def run_with_mcp_tools(user_query: str):
server_params = StdioServerParameters(
command="python",
args=["web_search_server.py"]
)
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
# MCP Server のツール定義を取得
tools_result = await session.list_tools()
tools = tools_result.tools
# OpenAI 形式の tools パラメータに変換
openai_tools = [
{
"type": "function",
"function": {
"name": tool.name,
"description": tool.description,
"parameters": tool.inputSchema
}
}
for tool in tools
]
# 標準 Function Calling フロー
response = client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": user_query}],
tools=openai_tools
)
# モデルがツールを呼び出そうとする場合、MCP Client 経由で実行
if response.choices[0].message.tool_calls:
for tc in response.choices[0].message.tool_calls:
result = await session.call_tool(
tc.function.name,
json.loads(tc.function.arguments)
)
# ... 結果をモデルに返すMCP のエンジニアリング価値
- 疎結合：ツール定義とサービス実装を分離し、それぞれ独立して反復可能
- 動的発見：Agent 起動時に最新のツールリストを自動取得、メンテナンスコストゼロ
- エコシステム効果：サードパーティが標準化された MCP Server を提供でき、Agent 開発者は MCP Client を統合するだけでよい
- マルチトランスポートプロトコル：stdio（ローカルプロセス通信）と HTTP/SSE（リモート通信）をサポート
リフレクションと自己修正
なぜリフレクションが必要か
大規模モデルが生成する内容は毎回使えるとは限らない。以下のようなことが起こり得る：
- コード内の変数名をこっそり「修正」して実行エラーを引き起こす
- 誤った「事実」に基づいて推論を続け、カスケードエラーを生む
- 長い出力の中で前の制約を忘れてしまう
リフレクション（Reflection）の中核的考え方は：モデルに自身が生成した完全な内容を再検討・評価する機会を与え、それによって誤りを発見し修正させることである。
自己フィードバックの二つのパターン
パターン一：単一ステップ指示型リフレクション
一回の呼び出しで、Prompt を通じてモデルに回答生成と同時にリフレクションを行うよう指示する：
prompt_with_reflection = """
## タスク
1. 以下のコースの言語表現を推敲し、推敲後の全文内容を出力してください。
2. 推敲後の内容をリフレクションしてください：
- 執筆規範に適合しているか
- 言語表現以外に、他の内容が意図せず変更されていないか
リフレクション結果と修正提案を出力してください。
3. 提案に照らしてコースを修正し、修正後の全文内容を出力してください。
## コース初稿
{original_content}
"""利点：実装がシンプルで、一回の呼び出しで完結する。 欠点：モデルが同じ思考バイアスを帯びたまま自己検証しやすく、「自己正当化」に陥る。
パターン二：二段階「生成-フィードバック」
生成と審査を二回の独立した呼び出しに分離する：
def generate_and_review(content: str) -> str:
# 第一段階：生成
draft_resp = client.chat.completions.create(
model="gpt-4",
messages=[{
"role": "system",
"content": "あなたはコース執筆者です。以下の内容を推敲し、より魅力的にしてください。"
}, {
"role": "user",
"content": content
}]
)
draft = draft_resp.choices[0].message.content
# 第二段階：審査（異なる system prompt を使用！）
review_resp = client.chat.completions.create(
model="gpt-4",
messages=[{
"role": "system",
"content": """あなたは厳格な技術審査員です。
【元の内容】と【推敲後の内容】を比較してください：
- 文案表現のみが修正され、コードなどの技術的内容が完全に一致している → 「通過」と回答
- 技術的内容が修正されている → 「不通過」と回答し、具体的な位置を指摘"""
}, {
"role": "user",
"content": f"元の内容:\n{content}\n\n推敲後の内容:\n{draft}"
}]
)
# 第三段階：不通過の場合、審査結果を作成 Agent にフィードバックして修正
review = review_resp.choices[0].message.content
if "不通過" in review:
# ... フィードバックを作成 Agent に戻して修正
pass
return draft中核的利点：審査 Agent の視点が執筆 Agent と異なるため、役割バイアスを回避できる。さらに複数の専門審査 Agent — 事実審査、論理審査、スタイル審査、安全審査 — を設置することも可能である。
外部フィードバック
自己フィードバックには本来的な限界がある：モデルは内容が実環境で正しいかどうかを検証できない。外部フィードバックの考え方は、生成結果を実環境で実行し、客観的事実で検証することである。
def generate_code_and_validate(spec: str) -> str:
# 第一段階：コード生成
code_resp = client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": f"以下の要件に基づいて Python コードを書いてください:\n{spec}"}]
)
code = extract_code(code_resp.choices[0].message.content)
# 第二段階：外部実行検証
import subprocess, tempfile
with tempfile.NamedTemporaryFile(suffix=".py", mode="w") as f:
f.write(code)
f.flush()
result = subprocess.run(
["python", f.name],
capture_output=True,
text=True,
timeout=30
)
# 第三段階：エラーをモデルにフィードバックして修正
if result.returncode != 0:
fix_prompt = f"""以下のコードの実行でエラーが発生しました：
コード:
{code}
エラー情報:
{result.stderr}
コードを修正し、完全な修正バージョンを出力してください。"""
fix_resp = client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": fix_prompt}]
)
return fix_resp.choices[0].message.content
return code外部フィードバックの適用シーン：
- コード実行検証：コードインタプリタでコードを実行し、ランタイムエラーを捕捉
- JSON Schema 検証：Pydantic などのライブラリで構造化出力を検証
- 数値計算検証：電卓ツールで数学的結果を検証
- 可視化レンダリング検証：グラフ生成後にモデルにレンダリング結果を「見せて」視覚検査
Plan & Execute パターン
なぜ明示的な計画が必要か
Agent に複雑なタスクを直接実行させると、以下の問題がよく発生する：
- 忘却：後続ステップの処理中に前の制約を「忘れる」
- カスケードエラー：初期の誤りが後続の推論の基礎となり、誤りが指数関数的に拡大
- 構造の見落とし：モデルは線形処理を好む傾向があり、タスクの並列/依存関係を識別できない
Plan & Execute パターンの中核思想：先に計画し後で実行する — まず完全な行動計画を策定し、審査で確認した後に段階的に実行する。
Plan Mode の実装
from typing import List
from pydantic import BaseModel
class PlanStep(BaseModel):
step_id: int
description: str
dependencies: List[int] = [] # 依存するステップ ID
tool: str = "" # 使用するツール
expected_output: str = "" # 期待される成果物
class ExecutionPlan(BaseModel):
goal: str
steps: List[PlanStep]
def plan_and_execute(task: str) -> str:
# Phase 1: 計画を策定
plan_prompt = f"""
あなたはプロジェクト計画の専門家です。以下のタスクに対して詳細な実行計画を策定してください。
要件：
1. タスクを具体的なステップに分解
2. ステップ間の依存関係を明示
3. 各ステップの期待される成果物を説明
タスク: {task}
JSON 形式で計画を出力してください。"""
plan_resp = client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": plan_prompt}],
response_format={"type": "json_object"}
)
plan = ExecutionPlan.model_validate_json(
plan_resp.choices[0].message.content
)
# Phase 2: 依存関係に従って実行
results = {}
executed = set()
while len(executed) < len(plan.steps):
for step in plan.steps:
if step.step_id in executed:
continue
# 依存がすべて実行済みか確認
if all(dep in executed for dep in step.dependencies):
# ステップを実行
result = execute_step(step, results)
results[step.step_id] = result
executed.add(step.step_id)
# Phase 3: 結果を集約
return summarize_results(plan, results)固定ワークフロー：Pipeline パターン
タスクのステップが確定的で再現可能な場合、パイプラインとして固定化すべきである：
Input → Step 1 → Step 2 → Step 3 → ... → Outputclass Pipeline:
"""固定パイプライン：各ステップの出力が次のステップの入力となる"""
def __init__(self):
self.steps = []
def add_step(self, name: str, func):
self.steps.append({"name": name, "func": func})
def run(self, input_data):
result = input_data
for step in self.steps:
print(f" [実行] {step['name']}")
result = step["func"](result)
return result
# 例：ドキュメント処理パイプライン
pipeline = Pipeline()
pipeline.add_step("PDF解析", parse_pdf_to_text)
pipeline.add_step("テキストクレンジング", clean_text)
pipeline.add_step("セクション分割", split_sections)
pipeline.add_step("ベクトル化", vectorize_chunks)
pipeline.add_step("インデックス保存", store_to_vectordb)
pipeline.run("document.pdf")ワークフローオーケストレーションパターン
五つの中核ワークフローパターン
複雑なタスクでは、Agent ノードを特定のトポロジーで組織化する必要がある。以下が五つの中核パターンである：
分岐選択（Branching/Router）
入口ノードでタスクタイプを判断し、異なる処理パスに振り分ける。
def router_agent(user_input: str):
"""意図に応じて異なる処理パイプラインに振り分け"""
classify_prompt = f"""
以下のユーザー要求のタイプを分析し、一語のみ返してください:
- code_review: コードのチェック/検証
- style_review: 言語の推敲/最適化
- fact_check: 事実/概念の正確性検証
要求: {user_input}
タイプ:"""
intent = client.chat.completions.create(
model="gpt-4o-mini", # 軽量モデルでコスト削減
messages=[{"role": "user", "content": classify_prompt}],
temperature=0
).choices[0].message.content.strip()
pipelines = {
"code_review": code_review_pipeline,
"style_review": style_review_pipeline,
"fact_check": fact_check_pipeline
}
return pipelines.get(intent, default_pipeline)(user_input)並列実行（Parallel）
相互に依存しないサブタスクを同時に配布し、最後に集約する。
┌→ コードチェック ──┐
ユーザー入力 → Split ─┼→ 事実審査 ──┼→ Merge → 集約レポート
└→ スタイル審査 ──┘import asyncio
async def parallel_review(notebook_content: str):
"""コース内容に対して三つの審査を並列実行"""
tasks = [
asyncio.create_task(check_code(notebook_content)),
asyncio.create_task(check_facts(notebook_content)),
asyncio.create_task(check_style(notebook_content))
]
code_result, fact_result, style_result = await asyncio.gather(*tasks)
# 集約
return generate_summary_report(code_result, fact_result, style_result)混合専門家（Mixture-of-Agents, MoA）
複数の異なるモデルが同一タスクを処理し、集約器が最適な結果を統合する。
┌→ モデル A (推論が得意) ──┐
ユーザー質問 → Split ─┼→ モデル B (創造性が得意) ──┼→ Aggregator → 最適回答
└→ モデル C (正確性が得意) ──┘MoA の中核的発見はモデルの「協調性」（Collaborativeness）である：あるモデルが他のモデルの出力を参照できる場合、多くの場合より高品質な応答を生成できる。
def mixture_of_agents(task: str):
"""MOA 実装：多モデル + 集約"""
# 第一層：提案者が並列生成
proposers = ["gpt-4", "claude-3-opus", "gemini-pro"]
proposals = []
for model in proposers:
resp = client.chat.completions.create(
model=model,
messages=[{"role": "user", "content": task}]
)
proposals.append(resp.choices[0].message.content)
# 第二層：集約器が統合
aggregator_prompt = f"""
以下は同一問題に対する {len(proposals)} 件の回答です。それらの長所を統合し、
最適な回答を一つ生成してください。
{format_proposals(proposals)}
統合回答:"""
final = client.chat.completions.create(
model="gpt-4", # 最強モデルで集約
messages=[{"role": "user", "content": aggregator_prompt}]
)
return final.choices[0].message.content人機協調（Human-in-the-Loop, HITL）
重要なノードで人間による審査を導入し、「AI 実行 → 人間承認 → AI 継続」のループを形成する。
def hitl_workflow(task: str):
"""人機協調ワークフロー"""
plan = generate_plan(task)
print(f"実行計画:\n{format_plan(plan)}")
approval = input("この計画を承認しますか？(y/n): ")
if approval.lower() != 'y':
return "タスクはキャンセルされました"
for step in plan.steps:
result = execute_step(step)
print(f"ステップ {step.step_id} 完了: {result['summary']}")
if step.get("requires_review"):
review = input(f"ステップ結果を審査してください (approve/modify/reject): ")
if review == "reject":
print("ステップが却下されました。再実行します...")
result = execute_step(step, feedback=review)
return generate_final_output()パターン選択方法論
| パターン | 適用シーン | 不適シーン |
|---|---|---|
| Pipeline | フローが固定的、ステップが線形 | 動的な意思決定が必要なタスク |
| Branching | 複数タイプの入力に異なる処理が必要 | 複数側面の同時処理が必要 |
| Parallel | サブタスクが相互独立、効率重視 | 依存連鎖のあるタスク |
| MoA | 高品質要求、創造的タスク | コスト重視の通常タスク |
| HITL | 高リスク意思決定、コンプライアンス要求 | 低遅延要求のリアルタイムシステム |
| Plan & Execute | フローが多変、探索が必要な新規タスク | 高度に反復的な確定的タスク |
ベストプラクティスは「探索-固定化」混合パターンである：まず Plan & Execute で最適案を探索し、次に Pipeline に固定化して大規模生産に用いる。
階層型協調パターン
Leader-Worker アーキテクチャ
階層型協調（Hierarchical/Team Leader Pattern）は最も直感的なマルチ Agent 協調パターンである。「プロジェクトマネージャー + チームメンバー」の組織構造を模倣する：
Leader Agent の責務：
- トップレベルタスクの受信と理解
- サブタスクへの分解と適切な Worker への割り当て
- 全体進捗の追跡
- Worker 成果の集約
Worker Agent は各々特定領域の専門性を持ち、割り当てられたサブタスクの実行に専念する。
Handoff による階層型協調の実装
class LeaderWorkerSystem:
"""階層型協調システムの実装"""
def __init__(self):
self.workers = {
"instructional_designer": self._create_worker(
"あなたは教育デザイナーです。コース概要と学習パスの設計を得意とします。"
),
"data_scientist": self._create_worker(
"あなたはデータサイエンティストです。Python データ分析のコードとケース作成を得意とします。"
),
"content_writer": self._create_worker(
"あなたはコンテンツ執筆者です。技術内容を生き生きとしたコース原稿に変換することを得意とします。"
)
}
self.leader = self._create_leader()
def _create_leader(self):
return {
"system_prompt": """あなたはコースプロジェクト主管です。あなたの責務は：
1. 要件を分析し、サブタスクに分解
2. タスクを適切な専門家に割り当て
3. 各専門家の成果を統合して完全なコースを構築
利用可能な専門家チーム：instructional_designer, data_scientist, content_writer""",
"tools": [
{
"type": "function",
"function": {
"name": "delegate_to_worker",
"description": "サブタスクを指定した専門家に割り当てる",
"parameters": {
"type": "object",
"properties": {
"worker": {
"type": "string",
"enum": list(self.workers.keys()),
"description": "タスクを担当する専門家"
},
"task": {
"type": "string",
"description": "具体的なサブタスクの説明"
}
},
"required": ["worker", "task"]
}
}
}
]
}
def run(self, project_brief: str) -> str:
"""完全なコース開発プロジェクトを実行"""
messages = [
{"role": "system", "content": self.leader["system_prompt"]},
{"role": "user", "content": project_brief}
]
# Leader ループ
while True:
response = client.chat.completions.create(
model="gpt-4",
messages=messages,
tools=self.leader["tools"]
)
msg = response.choices[0].message
if msg.content and not msg.tool_calls:
return msg.content # 最終集約出力
if msg.tool_calls:
messages.append(msg)
for tc in msg.tool_calls:
if tc.function.name == "delegate_to_worker":
args = json.loads(tc.function.arguments)
# Worker を呼び出してサブタスクを実行
worker_result = self._run_worker(
args["worker"], args["task"]
)
messages.append({
"role": "tool",
"tool_call_id": tc.id,
"content": worker_result
})
def _run_worker(self, worker_name: str, task: str) -> str:
worker = self.workers[worker_name]
resp = client.chat.completions.create(
model="gpt-4",
messages=[
{"role": "system", "content": worker},
{"role": "user", "content": task}
]
)
return resp.choices[0].message.content階層型協調の長所と短所
長所：
- 構造が明確で、各 Agent の責務がはっきりしている
- Leader が全体を掌握し、目標から逸脱しない
- 各 Worker が独立したコンテキストウィンドウを持ち、より集中できる
- タスクの並列割り当てをサポート
短所：
- Worker 間の直接コミュニケーションがなく、情報伝達に遅延/歪みが生じる
- Leader が単一障害点のボトルネックとなる
- 各モジュール結合後に全体の流暢さが欠ける可能性がある
黒板協調パターン
非中央集権的な共創
黒板パターン（Blackboard/Co-creation Pattern）は「専門家がホワイトボードを囲んでブレインストーミングする」作業方式を模倣する。中央集権的な調整者が存在せず、すべての Agent が平等に共有空間を読み書きする：
黒板パターンの実装
class BlackboardSystem:
"""黒板協調システム"""
def __init__(self, agents: dict, max_rounds: int = 3):
self.agents = agents
self.max_rounds = max_rounds
self.blackboard = [] # 共有空間
def run(self, problem: str) -> str:
# 問題を黒板に書き込む
self.blackboard.append({"source": "user", "content": problem})
for round_num in range(self.max_rounds):
print(f"\n=== 第 {round_num + 1} ラウンド ===")
new_contributions = []
# 全 Agent が並列に黒板を読み取って貢献
for name, agent_config in self.agents.items():
contribution = self._agent_contribute(
name, agent_config, self.blackboard
)
if contribution:
new_contributions.append({
"source": name,
"content": contribution
})
# 新たな貢献を黒板に書き込む
self.blackboard.extend(new_contributions)
# 合意に達したか確認
if self._check_consensus():
break
return self._synthesize_final_answer()
def _agent_contribute(self, name, config, blackboard):
"""各 Agent が黒板を読んだ後、自身の考えを貢献"""
board_text = self._format_blackboard(blackboard)
prompt = f"""あなたは{config['role']}です。
現在の共有黒板上の内容:
{board_text}
既存の議論に基づいて、あなたの見解、補足、疑問、または新しいアイデアを提案してください。
既存の案がすでに完成されている場合は、同意を表明し理由を説明しても構いません。"""
resp = client.chat.completions.create(
model="gpt-4",
messages=[
{"role": "system", "content": config["system_prompt"]},
{"role": "user", "content": prompt}
]
)
return resp.choices[0].message.content
def _format_blackboard(self, blackboard):
return "\n\n".join([
f"[{entry['source']}]: {entry['content']}"
for entry in blackboard
])
def _check_consensus(self):
"""黒板上の最新の貢献が合意形成されたか確認"""
# 合意検出ロジックを実装
pass
def _synthesize_final_answer(self):
"""黒板の内容から最終案を合成"""
pass黒板パターン vs 階層パターン
| 次元 | 階層パターン | 黒板パターン |
|---|---|---|
| 制御方式 | 中央集権的（Leader が掌握） | 非中央集権的（平等参加） |
| 通信方式 | スター型（Leader↔Worker） | 全結合（全 Agent↔黒板） |
| 意思決定機構 | Leader が裁決 | 創発的合意 |
| 適するタスク | 目標が明確、分解可能 | 開放的な探索、集合知が必要 |
| 効率 | 高（並列 + 制御可能） | 低め（複数ラウンドの討論） |
| 創造性 | 限定的（Leader の視点に制約） | 高（思想の衝突が新しいアイデアを生む） |
| コスト | 中等 | 高（全 Agent が毎ラウンド参加） |
協調パターン選択方法論
現実世界から学ぶ
優れたマルチ Agent システムの設計は、現実世界のチーム協調の観察と抽出に由来する。抽象的なパターン名称を暗記するより、業務に入り込み、人間の専門家チームが類似のタスクをどのように完了するかを観察する方が有用である。
観察の三次元：
混合パターン設計
実際のプロジェクトで単一パターンのみを使用することは稀である。混合設計が一般的である：
┌─────────────┐
│ Leader │ ← 階層パターン
└──────┬──────┘
┌───────────────┼───────────────┐
▼ ▼ ▼
┌──────────┐ ┌──────────┐ ┌──────────┐
│ 教育デザイナー │ │ コンテンツ執筆者 │ │ 審査 Leader│
│ Worker │ │ Worker │ └─────┬────┘
└──────────┘ └──────────┘ │
┌────────┼────────┐
▼ ▼ ▼
┌──────┐ ┌──────┐ ┌──────┐
│コード検 │ │事実検 │ │スタイル検 │ ← 並列パターン
└──────┘ └──────┘ └──────┘
│ │ │
└────────┼────────┘
▼
┌──────────┐
│ 統合レポート │
└──────────┘コスト意識
マルチ Agent システムの Token 消費は通常、単一 Agent の 3-5 倍である。設計時には以下のトレードオフが必要である：
def estimate_cost(num_agents: int, avg_tokens_per_agent: int,
rounds: int = 1, price_per_1k: float = 0.01):
"""マルチ Agent システムの Token コストを推定"""
total_tokens = num_agents * avg_tokens_per_agent * rounds
return total_tokens * price_per_1k / 1000
# 例：5 Agent、各 2000 tokens、3 ラウンドの黒板討論
cost = estimate_cost(5, 2000, 3)
print(f"推定コスト: ${cost:.2f}")
# 実際の数値はより高くなる可能性がある（黒板内容の重複伝送も含まれるため）短期記憶管理
無状態性：問題の根源
大規模言語モデルは本質的に**無状態（Stateless）**である。各 API 呼び出しは独立している — 前の対話ラウンドの内容、あなたの嗜好、あるいは以前に達した合意を覚えていない。
# この二回の呼び出しは互いに完全に独立
response1 = client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": "私の名前は張三です"}]
)
# 応答: "こんにちは張三さん！どのようなご用件でしょうか？"
response2 = client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": "私の名前は何ですか？"}]
)
# 応答: "申し訳ありませんが、以前に対話したことがないため、あなたのお名前はわかりません。"解決策：対話履歴リストを維持し、毎回の呼び出しで完全な履歴を送信する。
class ConversationBuffer:
"""最もシンプルな短期記憶：完全な対話履歴を保存"""
def __init__(self, system_prompt: str = ""):
self.messages = []
if system_prompt:
self.messages.append({"role": "system", "content": system_prompt})
def chat(self, user_input: str) -> str:
self.messages.append({"role": "user", "content": user_input})
response = client.chat.completions.create(
model="gpt-4",
messages=self.messages
)
reply = response.choices[0].message.content
self.messages.append({"role": "assistant", "content": reply})
return replyコンテキストウィンドウの圧力
対話ターン数の増加に伴い、完全履歴方式は三つの致命的な問題に直面する：
- コンテキストウィンドウ超過：履歴の長さがモデルの制限を超える → プログラムエラー
- コスト制御不能：毎回の呼び出しで全履歴を再送信 → Token 消費が線形増加
- 注意の希釈：長いコンテキストでは、モデルの中間部分の情報処理能力が著しく低下
三つの記憶管理戦略
戦略一：固定ウィンドウ打ち切り（Context Truncation）
直近の N ターンの対話または N 個の Token のみを保持する。
class TruncationMemory:
def __init__(self, max_tokens: int = 4000):
self.max_tokens = max_tokens
self.messages = []
def add_and_truncate(self, role: str, content: str):
self.messages.append({"role": role, "content": content})
# 最も古いメッセージから削除し、総 token 数を制限内に収める
while self._total_tokens() > self.max_tokens:
self.messages.pop(0) # 最も古い非 system メッセージを削除
def _total_tokens(self):
return sum(count_tokens(m["content"]) for m in self.messages)利点：実装が極めてシンプルで、計算オーバーヘッドが小さい。 欠点：重要な情報が初期の対話にある場合、打ち切られると Agent は「記憶喪失」になる。
戦略二：ローリングサマリー（Rolling Summary）
忘れる前に、まず要点を抽出する。
対話履歴: [msg1, msg2, msg3, msg4, msg5, msg6, msg7, msg8]
↓ 前半を圧縮
[要約(m1-m4), msg5, msg6, msg7, msg8]
↓ 継続して圧縮
[要約(m1-m6), msg7, msg8]class RollingSummaryMemory:
def __init__(self, summary_trigger_tokens: int = 3000):
self.summary_trigger = summary_trigger_tokens
self.messages = []
self.summary = ""
def add_message(self, role: str, content: str):
self.messages.append({"role": role, "content": content})
if self._total_tokens() > self.summary_trigger:
self._compress()
def _compress(self):
"""前半の対話を要約に圧縮"""
split_point = len(self.messages) // 2
to_compress = self.messages[:split_point]
remaining = self.messages[split_point:]
compress_prompt = f"""
以下の対話履歴を簡潔な要約にまとめ、重要な情報を保持してください：
対話:
{format_messages(to_compress)}
要約:"""
resp = client.chat.completions.create(
model="gpt-4o-mini", # 軽量モデルで要約
messages=[{"role": "user", "content": compress_prompt}]
)
self.summary = resp.choices[0].message.content
# 圧縮されたメッセージを要約で置き換え
self.messages = [
{"role": "system", "content": f"対話履歴要約:\n{self.summary}"}
] + remaining
def _total_tokens(self):
return sum(count_tokens(m["content"]) for m in self.messages)利点：長さを圧縮しながら中核情報を保持し、長期的な一貫性を維持する。 欠点：追加の API 呼び出しコストがかかり、要約の品質が後続の対話に直接影響する。
戦略三：ベクトル化検索（Vector-based Retrieval）
最もスマートな方式：対話履歴をベクトルデータベースに保存し、必要に応じて最も関連する記憶を検索する。
class VectorBasedMemory:
def __init__(self):
self.conversations = [] # 完全な対話記録
self.embeddings = [] # 各対話ターンのベクトル
self.embed_model = "text-embedding-3-small"
def store_conversation(self, user_msg: str, assistant_msg: str):
"""対話を一ターン保存しベクトル化"""
conversation_text = f"User: {user_msg}\nAssistant: {assistant_msg}"
self.conversations.append(conversation_text)
vec = client.embeddings.create(
model=self.embed_model,
input=conversation_text
)
self.embeddings.append(vec.data[0].embedding)
def retrieve_relevant(self, current_query: str, top_k: int = 5):
"""現在の質問に最も関連する履歴対話を検索"""
query_vec = client.embeddings.create(
model=self.embed_model,
input=current_query
).data[0].embedding
# 類似度を計算
similarities = [
np.dot(query_vec, mem_vec) /
(np.linalg.norm(query_vec) * np.linalg.norm(mem_vec))
for mem_vec in self.embeddings
]
top_indices = np.argsort(similarities)[-top_k:][::-1]
return [self.conversations[i] for i in top_indices]利点：コンテキストウィンドウの長さ制限から根本的に脱却し、意味的に正確にマッチングする。 欠点：システムの複雑さが最も高く、Embedding モデルとベクトルデータベースを導入する必要がある。
戦略選択ガイド
| シーン | 推奨戦略 |
|---|---|
| 雑談ボット | 固定ウィンドウ打ち切り（シンプルで効果的） |
| カスタマーサービスQ&A（情報価値が時間とともに急速に減衰） | 固定ウィンドウ打ち切り |
| 長編コンテンツ制作 / プロジェクト計画 | ローリングサマリー |
| パーソナライズドアシスタント / 長期インタラクション | ベクトル化検索 |
| ベストプラクティス | 混合使用：サマリー + ベクトル検索 |
長期記憶とベクトルストレージ
受動的コンテキストから能動的記憶管理へ
真に知的な Agent は、処理済みのコンテキストを受動的に受け取るだけでなく、自身の記憶を能動的に管理できるべきである — いつ何を記憶すべきか、いつ何を思い出すべきかを自ら決定する。
そのためには Agent に二つの中核ツールを提供する必要がある：
# Agent が利用可能な記憶管理ツール
memory_tools = [
{
"type": "function",
"function": {
"name": "record_to_memory",
"description": "重要な情報を長期記憶に保存する。ユーザーが明示的に嗜好を表明した場合、重要な情報を提供した場合、または重要な意思決定を完了した場合に使用する。",
"parameters": {
"type": "object",
"properties": {
"content": {
"type": "string",
"description": "記憶すべき内容"
},
"category": {
"type": "string",
"enum": ["preference", "fact", "decision", "context"],
"description": "記憶カテゴリ"
}
},
"required": ["content"]
}
}
},
{
"type": "function",
"function": {
"name": "retrieve_from_memory",
"description": "長期記憶から関連情報を検索する。ユーザーの嗜好、過去の意思決定、または以前に議論した内容を思い出す必要がある場合に使用する。",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "検索クエリ"
},
"category": {
"type": "string",
"description": "オプション。検索する記憶カテゴリを限定"
}
},
"required": ["query"]
}
}
}
]短期記憶 vs 長期記憶
┌─────────────────────────────────────────────────────┐
│ 記憶システムアーキテクチャ │
├─────────────────────────────────────────────────────┤
│ │
│ 短期記憶 (Short-term Memory) │
│ ┌───────────────────────────────────────────────┐ │
│ │ ストレージ：対話バッファ (messages list) │ │
│ │ 管理：打ち切り / 要約 │ │
│ │ ライフサイクル：現在のセッション │ │
│ │ 責務：セッションの一貫性維持、「今何を話しているか」を記憶 │ │
│ └───────────────────────────────────────────────┘ │
│ │
│ 長期記憶 (Long-term Memory) │
│ ┌───────────────────────────────────────────────┐ │
│ │ ストレージ：ベクトルデータベース + メタデータストア │ │
│ │ 管理：ベクトル化検索 + ツール化呼び出し │ │
│ │ ライフサイクル：セッション横断 │ │
│ │ 責務：重要な情報の永続化、セッション横断的知的検索 │ │
│ └───────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────┘記憶管理のベストプラクティス
-
選択的に記憶する：記憶は多ければ多いほど良いわけではない。低価値な情報は後続の検索を妨害する。書き込みの許可機構を確立する — ユーザーが明示的に要求した場合、または情報の重要性が閾値を超えた場合のみ書き込む。
-
継続的ガバナンス：記憶は動的なデータ資産である。定期的に古い情報をクリーンアップし、重複エントリを統合し、事実の正確性を検証する。ユーザーが記憶を管理するインターフェース（閲覧、修正、削除）を提供する。
-
シーンに応じた適用：シーンによって記憶要件は異なる。コースドキュメントのワークフローでは個人化された嗜好を記録すべきではない。製品の事実情報（API パラメータ、機能制限など）については記録し定期的に有効性を審査すべきである。
Skill システム設計
Prompt から Skill への進化
入念に作成された Prompt の価値は高いが、それは現在のセッションでのみ有効である。セッションが終われば、知識は散逸する。Skill は Prompt を再利用可能、バージョン管理可能、チーム共有可能な専門知識モジュールへとアップグレードするものである。
進化のパス：
その場限りの Prompt ("このコースを審査して")
↓ 問題：毎回書き直す必要があり、基準が一貫しない
固定 Prompt (チャット履歴に保存)
↓ 問題：散在し、見つけにくく、協調できない
独立ファイル (course-review.md)
↓ 問題：ファイルが肥大化し、保守が困難
ナレッジベースディレクトリ (course-review/)
↓ 問題：依然として Agent に手動で何をすべきか伝える必要がある
Skill (SKILL.md + リソースファイル + スクリプト)Skill の構造
規範的な Skill は YAML frontmatter + Markdown body で構成される：
---
name: course-review
description: |
コース内容の技術的正確性、コードの正確性、教育品質を審査する。
ユーザーが既存のコースやトレーニング教材の審査、監査、評価を要求した場合にこのスキルを使用する。
---
# コース審査スキル
## 審査フロー
1. Notebook のディレクトリ構造を抽出し、全体の章構成を理解する
2. 章ごとに段落を確認し、以下の観点に照らしてチェックする：
- コード実行可能性（[code-quality.md](code-quality.md) 参照）
- 内容の正確性（[content-accuracy.md](content-accuracy.md) 参照）
- 解説スタイル（[style-guide.md](style-guide.md) 参照）
- 古い API（[outdated-api.md](outdated-api.md) 参照）
3. 審査結果を集約し、出力形式に従ってレポートを生成する
## アンチパターンリスト
- ❌ コード内の変数名、APIバージョン番号を変更しない
- ❌ いかなるチェック項目もスキップしない
- ❌ 審査プロセス中に新しい技術概念を導入しない
## 出力形式
各チェック項目について：通過/不通過/要人手確認 + 位置 + 修正提案ディレクトリ構造：
course-review/
├── SKILL.md # メイン指示エントリポイント
├── code-quality.md # コード実行可能性チェック項目
├── content-accuracy.md # 事実正確性チェック項目
├── style-guide.md # 解説スタイルの正例と反例
├── outdated-api.md # 古い API 対照表
└── scripts/
├── extract_toc.py # Notebook ディレクトリ抽出
└── validate_code.py # コード検証の自動実行Skill vs RAG
多くの人が Skill と RAG を混同する。それらの中核的区别は：
| RAG | Skill | |
|---|---|---|
| 解決する問題 | 「モデルがある事実を知らない」 | 「モデルがどうすればよいかわからない」 |
| 情報タイプ | 事実的知識（ドキュメント内容、製品パラメータ） | 手続き的知識（フロー、基準、判断ルール） |
| トリガー方式 | 検索後にコンテキストに注入 | 選択後に展開（Agent が活性化を判断） |
| ロード方式 | 検索結果を一度に注入 | 漸進的開示（必要に応じて段階的にサブファイルをロード） |
| ライフサイクル | 毎回のクエリで個別検索 | セッション横断で永続化、バージョン管理可能 |
高品質な Skill 作成の五段階法
第一段階：作る価値があるか判断する
├─ このタスクには「専門家の直感」があるか？（専門家はできるが初心者は境界条件を見落としがち）
├─ このタスクは十分に複雑か？（GUI で3ステップ以内に完了できるなら、作らない）
└─ このタスクは繰り返し実行されるか？（一度だけなら、作らない）
第二段階：何を書くべきか抽出する
├─ 単なる手順ではなく専門家の決定木を抽出する
├─ アンチパターンチェックを注入する（「絶対に踏んではいけない落とし穴」）
├─ Template パターン：標準化された出力テンプレートを提供する
└─ Examples パターン：テキスト説明の代わりに例示を使用する
第三段階：指示を適切に書く
├─ 簡潔に：一文一文がその token コストに値するものでなければならない
├─ 自由度のマッチング：制約の度合いをタスクリスクに合わせる
│ ├─ 低自由度（データベースマイグレーション）：正確なスクリプト
│ ├─ 中自由度（レポート生成）：疑似コード/パラメータ化
│ └─ 高自由度（コードレビュー）：テキスト指示
└─ 漸進的開示：メインファイルは簡潔に保ち、詳細は必要に応じてロード
第四段階：ツールを適切に装備する
├─ ワークフロー：追跡可能な Checklist
├─ フィードバックループ：実行→チェック→修正→繰り返し
├─ 高リスク操作：実行前に計画を検証する
└─ AI フレンドリーなスクリプト：構造化状態 + 修正の手がかり + 優雅なデグラデーション + 冪等安全性
第五段階：検証と反復
├─ フェーズ1：評価ベースラインを確立する（評価があってから Skill を書く）
├─ フェーズ2：Skill を抽出する（AI に繰り返し提供した情報を要約させる）
└─ フェーズ3：デュアル Agent テスト反復（設計者 vs 使用者）Skill as Code
Skill をコードとして扱い、ソフトウェアエンジニアリングの方法論の恩恵を受ける：
- バージョン管理：Skill ディレクトリを Git に含め、各修正に履歴記録を残す
- コードレビュー：新規/修正 Skill には PR レビューが必要
- CI/CD：Skill 変更時に評価パイプラインをトリガーし、退行がないことを確認
- コミュニティ共有：オープンソースライブラリのように共有・再利用する
漸進的情報開示と Skill as Code
漸進的情報開示の設計哲学
従来の Prompt エンジニアリングが直面する中核的矛盾：情報が多すぎる → コンテキストが混雑し、注意が希釈される。情報が少なすぎる → Agent が意思決定を支える十分な知識を持てない。
漸進的情報開示（Progressive Disclosure） はこの矛盾を解決する設計パターンである：
第 1 層：Skill リスト（Agent 起動時に可視）
↓ Agent がある Skill の活性化を選択
第 2 層：SKILL.md（Skill 活性化後にロード）
↓ Agent があるステップを実行
第 3 層：サブファイル（必要に応じて対応リソースをロード）
↓ ある技術的詳細を検証する必要がある
第 4 層：スクリプト実行（決定論的な結果を提供）設計原則：
- フラットな構造を保ち、深いネスト参照を避ける
- SKILL.md がすべてのリソースファイルに直接リンクし、「ワンステップ」で到達可能にする
- すべての内容を SKILL.md に詰め込まない — それは単なる入口である
AI フレンドリーなスクリプト設計
ツールスクリプトの出力品質は Agent のパフォーマンスに直接影響する。四つの重要な原則：
1. 構造化された状態フィードバック — フリーテキストではなく JSON を出力する：
# ❌ 悪い出力
print("Error: exit code 1")
print("Could not find module 'openpyxl'")
# ✅ AI フレンドリーな出力
print(json.dumps({
"status": "failed",
"error_code": "MODULE_NOT_FOUND",
"missing_module": "openpyxl",
"fix_hint": "pip install openpyxl を実行して不足している依存関係をインストールしてください",
"fallback_available": True,
"affected_cells": [42, 43, 44]
}))2. エラーメッセージに修正の手がかりを含める — 「何が起きたか」だけでなく「どう修正するか」を Agent に伝える。
3. クラッシュではなく優雅なデグラデーション — 可能な場合はデフォルト値を提供して継続する。
4. 冪等性と安全性 — 副作用なく繰り返し実行できる：
| シーン | 非冪等（危険） | 冪等（安全） |
|---|---|---|
| ファイル書き込み | 毎回内容を追記 | まずクリアしてから書き込み |
| データベース操作 | 毎回 INSERT | UPSERT を使用 |
| API 呼び出し | 毎回新リソースを作成 | 冪等キーを使用 |
評価フレームワーク設計
なぜ「なんとなく良い感じ」は当てにならないか
Agent の最適化を主観的な感覚だけに頼ると、以下の問題を引き起こす：
- 定量化困難：「なんとなく良い」はエンジニアリング上の意思決定根拠にならない
- 基準の欠如：異なるテスター、異なる時点での評価基準がずれる可能性がある
- 再現不可能：体系的なリグレッションテストができず、新たな変更が旧機能を壊していないことを確認できない
評価駆動開発（Evaluation-Driven Development） は評価を開発プロセスの末端から中核的位置に引き上げる：
┌──────────────────┐
│ 評価 = 品質の物差し │
└────────┬─────────┘
│
┌─────────────┼─────────────┐
▼ ▼ ▼
測定できて初めて フィードバックが早く正確なほど 製品能力の
改善できる 改善が効率的になる 上限を決めるエンドツーエンド評価（End-to-End Evaluation）
エンドツーエンド評価は最終出力に注目し、「この Agent はユーザーにとって使いやすいか？」に答える。
評価指標は二種類に分類される：
| タイプ | 説明 | 例 |
|---|---|---|
| 客観的指標 | コードルールで直接判断可能 | コードが実行可能か、形式が Schema に合致するか、文字数が範囲内か |
| 主観的指標 | 意味と品質の判断を含む | 内容の正確性、教育的有効性、言語スタイルが規定に適合しているか |
ホワイトボックス評価（White-box Evaluation）
Agent のプロセスが複雑化すると、エンドツーエンド評価では具体的な問題を特定できない。ホワイトボックス評価はシステム内部に深く入り込み、重要なコンポーネントごとに個別の評価体系を設計することを主張する。
エンドツーエンド評価 (最終出力のみを見る):
Agent 全体 → スコア 4.2/5 → しかしどこに問題があるかわからない
ホワイトボックス評価 (中間工程をチェック):
┌────────────┐ ┌────────────┐ ┌────────────┐
│ 概念説明 │ │ コード生成 │ │ スタイル調整 │
│ スコア 3.1/5│ │ スコア 4.8/5│ │ スコア 4.5/5│
└────────────┘ └────────────┘ └────────────┘
↑ ボトルネックはここ！ホワイトボックス評価の中核的利点：
- 明確なシグナル：ノイズのない改善シグナル、真のボトルネックに集中
- 高速反復：単一コンポーネントのみをテストすればよく、フロー全体を実行する必要なし
- 精密な最適化：各変更の効果を正確に測定可能
LLM-as-Judge
大規模モデルを評価員として使う
別の大規模モデルに「評価専門家」を演じさせ、定義された指標と採点細則に基づいて自動採点させる。
def llm_judge_evaluate(generated_output: str, criteria: dict):
"""LLM を評価員として使用"""
judge_prompt = f"""
あなたは専門のコース品質評価員です。以下の基準に従って採点してください：
評価基準:
{json.dumps(criteria, indent=2, ensure_ascii=False)}
評価対象内容:
{generated_output}
JSON 形式で評価結果を出力してください：
{{
"scores": {{
"accuracy": <1-5>,
"clarity": <1-5>,
"engagement": <1-5>
}},
"overall": <1-5>,
"comments": "総合評価"
}}"""
resp = client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": judge_prompt}],
response_format={"type": "json_object"}
)
return json.loads(resp.choices[0].message.content)LLM 評価員のバイアス
LLM を評価員として使用する場合、その固有のバイアスに警戒しなければならない：
| バイアスタイプ | 現象 | 緩和方法 |
|---|---|---|
| スタイルバイアス | 特定のコード/ライティングスタイルを好む | 評価基準を明確にし、「好み」に依存しない |
| 長さバイアス | 長い回答を「より完全」と見なす | 「簡潔性」を評価次元として含める |
| 「イエスマン」バイアス | 肯定的な評価を好む傾向 | 対照評価（A vs B どちらが良いか）を使用 |
| 位置バイアス | リスト内の特定位置の内容を好む傾向 | 評価対象内容の順序をランダムに入れ替える |
ベストプラクティス：初期段階で人間の専門家を用いて「ゴールドテストセット」を確立し、それを使って LLM 評価員をキャリブレーションする。定期的に人手によるサンプルチェックで自動評価の一貫性を検証する。
評価指標の分解
曖昧な評価目標を、一つずつチェック可能な具体的細則に分解する：
# 内容品質評価の分解例
評価次元: "内容品質"
細則:
- id: "pain_point"
description: "具体的な痛点を導入として始めているか？"
type: "boolean"
- id: "theory_depth"
description: "初歩的な解法の限界を明確に指摘し、中核理論を導出しているか？"
type: "boolean"
- id: "code_relevance"
description: "コード例は解説する理論と密接に関連し、十分に簡素化されているか？"
type: "boolean"
- id: "anti_pattern_check"
description: "「新スキル獲得おめでとう」のような軽薄な表現を避けているか？"
type: "boolean"評価駆動反復
評価閉ループ
評価は一度限りのものではなく、改善を持続的に駆動するエンジンである：
┌──────────────────────────────────┐
│ │
▼ │
┌─────────┐ ┌──────────┐ ┌─────────┐
│ MVP 構築 │ → │ 問題発見 │ → │ 指標抽出 │
└─────────┘ └──────────┘ └─────────┘
│
▼
┌─────────┐ ┌──────────┐ ┌─────────┐
│ 本番デプロイ │ ← │ リグレッション │ ← │ 最適化改善 │
└─────────┘ └──────────┘ └─────────┘
│
└────→ (ループ)業務専門家主導の評価基準
評価指標（特に主観的指標）は最も経験豊富な業務専門家が主導して策定しなければならない：
-
業務目標で参加を動機づける：「評価指標の定義を手伝ってください」ではなく、「この Agent はコース制作サイクルを2週間から3日に短縮し、同時に90%以上のユーザー満足度を維持するのを手伝います」と言う。
-
構造化ツールを提供して参入障壁を下げる：採点尺度テンプレート、事例アノテーションツール、「コースの良し悪しを3つの指標だけで判断するとしたら、どの3つを選びますか？」といった誘導的な質問。
-
継続的協調メカニズムを確立する：毎週の振り返り会議で、専門家がデータを見て、技術チームがシステムを調整し、共同で意思決定する。
評価の効率レバレッジ
すべての評価が完全自動・完全カバレッジである必要はない。最もシンプルな方法から始める：
Level 1: 手動サンプルチェック → 「コードをコピーして実行してみる」
Level 2: 自動化スクリプト → 「スクリプトを書いてバッチ実行する」
Level 3: 統合評価パイプライン → CI/CD 統合、PR ごとに自動トリガー
Level 4: 継続的モニタリング → 本番環境で重要指標をリアルタイム監視各 Level の投資対効果は異なる。初期段階では Level 1 の投資対効果が最も高い — 最も早く問題を発見でき、実装コストが最も低い。システムの成熟に伴い、徐々により高い Level へと進化させる。
モデルデプロイメント戦略
業務要件分析フレームワーク
大規模モデルアプリケーションを本番環境にリリースする際、最初のステップは技術選定ではなく要件分析である：
┌────────────────────────────────────────────┐
│ 業務要件分析マトリックス │
├────────────────────────────────────────────┤
│ │
│ 機能的要件 (何をするか): │
│ ├─ 自然言語処理 → 汎用 LLM │
│ ├─ コード生成 → コード最適化 LLM │
│ ├─ 数学推論 → 数学ファインチューニング LLM │
│ ├─ 視覚理解 → マルチモーダルモデル │
│ └─ 音声処理 → 音声モデル │
│ │
│ 非機能的要件 (どのように行うか): │
│ ├─ 性能：TTFT < 500ms, TPOT < 50ms │
│ ├─ コスト：単一呼び出し < $0.01 │
│ ├─ 安定性：99.9% 可用性 │
│ ├─ セキュリティ：コンテンツフィルタリング、プライバシー保護 │
│ └─ コンプライアンス：業界規制要件 │
│ │
└────────────────────────────────────────────┘モデル選定戦略
すべてのシーンで最大のモデルが必要なわけではない。モデル選定は「最小可用」原則に従う：
タスク複雑度
↑
│ ┌──────────────────────────┐
│ │ 大規模モデル (GPT-4, Claude) │
│ │ - 複雑な推論 │
│ │ - 多段階計画 │
│ │ - 創造的生成 │
│ └──────────────────────────┘
│ ┌──────────────────────────┐
│ │ 中規模モデル (GPT-4o-mini) │
│ │ - 意図認識 │
│ │ - 構造化抽出 │
│ │ - 要約生成 │
│ └──────────────────────────┘
│ ┌──────────────────────────┐
│ │ 小規模モデル / 蒸留モデル │
│ │ - テキスト分類 │
│ │ - キーワードマッチング │
│ │ - 形式検証 │
│ └──────────────────────────┘
└─────────────────────────────────→ 呼び出し頻度蒸留：小規模モデルに専門能力を習得させる
蒸留の中核思想：大規模モデルの判断能力を小規模モデルに「コピー」する。
教師モデル (GPT-4) 生徒モデル (0.6B パラメータ)
│ │
│ アノテーションデータを生成 │
├─────────────────────────→│
│ "要求の意図を理解せよ" │ 教師の行動パターンを学習
│ [入力→出力 ペア] │
│ │
│ 効果：小規模モデルが特定タスクで │
│ 教師モデルの水準に近づく │蒸留 vs ファインチューニング：
| ファインチューニング | 蒸留 | |
|---|---|---|
| データソース | 人手アノテーション | 教師モデルが生成 |
| データコスト | 高い | 低い（API 呼び出し費用） |
| データ規模 | 限定的 | 大規模生成可能 |
| 品質上限 | アノテーターに依存 | 教師モデルに依存 |
蒸留の三つのパス：
| パス | 必要なリソース | 適用シーン |
|---|---|---|
| データ合成蒸留（ブラックボックス） | API アクセスのみ | 構造化タスク、商用 API 教師 |
| 知識蒸留 KD（ホワイトボックス） | 教師モデルの重み | オープンソース教師、より高い精度が必要 |
| 推論圧縮 | 教師の推論軌跡 | 多段階推論タスク（DeepSeek-R1 など） |
推論最適化
性能最適化フレームワーク
LLM 推論最適化を四つの方向に分類する：
より速くリクエストを処理する
- モデル小型化：よりパラメータ数の少ないモデル変種を選択
- 量子化：INT4/INT8/FP16 量子化で計算リソース要件を削減
- プルーニング：冗長な重みを除去し、モデルの複雑さを低減
- 知識蒸留：大規模モデルのデータで小規模モデルを学習
量子化精度の比較:
FP32 (全精度) → ベースライン性能、最高の計算オーバーヘッド
FP16 (半精度) → ~2x 高速化、精度はほぼ無損失
INT8 → ~4x 高速化、わずかな精度損失
INT4 → ~8x 高速化、精度影響の慎重な評価が必要処理リクエスト数を減らす
- コンテキストキャッシュ（Context Cache）：複数ターン対話の共通プレフィックスをキャッシュし、重複計算を削減
- バッチ処理（Batching）：複数のリクエストを一つのバッチに統合し、ハードウェア利用率を向上
- 結果キャッシュ：高頻度の同じ質問の結果を直接キャッシュから返す
# コンテキストキャッシュの典型的な応用
# 複数ターン対話では、System Prompt + 履歴知識ドキュメントが共通プレフィックス
# 初回ターン：全量計算（フル料金）
# 後続ターン：キャッシュヒット部分は 20% 料金で課金Token の入出力を減らす
- 入力側：入力を簡素化し、冗長な情報を除去、長文ドキュメントは先に要約を生成
- 出力側：Prompt で簡潔な回答を誘導し、合理的な
max_tokensを設定
max_tokens の設計哲学：それはコンテンツ制御手段ではなく安全弁である。意味的に完全な簡潔な応答は Prompt で誘導すべきである。max_tokens はコスト制御の最後の防衛線としてより適している。
並列化処理
大規模モデル推論の本質は大規模行列演算である。CPU と GPU の違いを理解する：
| CPU | GPU | |
|---|---|---|
| コア数 | 少数の強力なコア (8-64) | 膨大な数のシンプルなコア (数千) |
| 適するタスク | 複雑なロジック、直列処理 | 大規模並列行列演算 |
GPU 並列化戦略：
- データ並列：データを分割して複数の GPU に割り当て
- モデル並列：モデルの異なる層を異なるデバイスに分散
- パイプライン並列：計算プロセスを段階に分割して順次実行
大規模モデルにデフォルトで依存しない
多くのシーンでは、よりシンプルな方法の方が効率的である：
| シーン | 代替案 |
|---|---|
| 標準確認メッセージ | ハードコードテンプレート + ランダム変種選択 |
| 有限選択肢の応答 | すべての可能結果を事前計算し、入力に応じてマッチング |
| データ表示 | LLM が説明文を生成する代わりにグラフ、表など従来の UI を使用 |
| キーワードマッチング | 意図認識段階ではまずキーワードでフィルタし、必要な場合のみ LLM を呼び出す |
安全ガードレール
大規模モデルが直面するセキュリティ脅威
大規模モデルアプリケーションは多層的なセキュリティ脅威に直面し、体系的な防御戦略が必要である：
防御戦略マトリックス
| 攻撃タイプ | 攻撃方法 | 防御措置 |
|---|---|---|
| プロンプトインジェクション | モデルを誘導してシステム指示を上書き | 内蔵安全ガードレール検出 + ユーザー入力とシステム指示の厳格な分離 |
| コマンドインジェクション | リクエストに悪意のあるコードを埋め込む | 実行前監査 + 最小権限 |
| プロンプト漏洩 | モデルを誘導して自身の System Prompt を出力させる | 安全ガードレールで探索パターンを識別 |
| ナレッジベース汚染 | 誤った情報を含むドキュメントをアップロード | ナレッジ登録承認プロセス + コンテンツ事前スキャン |
| モデル窃取 | 大量の API 呼び出しで学習データを収集 | API レート制限 + ボットトラフィック識別 |
| 悪意のある機能呼び出し | Agent を誘導して危険な操作を実行 | ツール呼び出し前監査 + サーキットブレーカー機構 |
安全ガードレールのエンジニアリング実装
class SafetyGuard:
"""多層安全ガードレール実装"""
def __init__(self):
self.blocked_keywords = set() # カスタム機密語
self.rate_limits = {} # レート制限記録
self.max_tool_calls = 10 # Agent 最大ツール呼び出し回数
self.dangerous_commands = { # 危険コマンドブラックリスト
"rm -rf", "DROP TABLE", "DELETE FROM",
"os.system", "subprocess.call", "eval("
}
def check_input(self, user_input: str) -> tuple[bool, str]:
"""入力セキュリティチェック"""
# 1. 機密語検出
for keyword in self.blocked_keywords:
if keyword in user_input.lower():
return False, f"入力に機密語が含まれています: {keyword}"
# 2. コマンドインジェクション検出
for dangerous in self.dangerous_commands:
if dangerous.lower() in user_input.lower():
return False, f"潜在的に危険な命令を検出: {dangerous}"
return True, "通過"
def check_tool_call(self, tool_name: str, args: dict) -> tuple[bool, str]:
"""ツール呼び出し前監査"""
# 1. 呼び出し頻度チェック
if self.rate_limits.get(tool_name, 0) >= self.max_tool_calls:
return False, f"ツール {tool_name} の呼び出し回数が上限を超えています"
# 2. パラメータ安全性チェック
args_str = json.dumps(args).lower()
for dangerous in self.dangerous_commands:
if dangerous.lower() in args_str:
return False, f"ツールパラメータに危険な命令が含まれています: {dangerous}"
self.rate_limits[tool_name] = self.rate_limits.get(tool_name, 0) + 1
return True, "通過"
def check_output(self, output: str) -> tuple[bool, str]:
"""出力コンテンツ審査"""
# 出力すべきでない機密情報パターンを検出
sensitive_patterns = [
r'\b\d{17}[\dXx]\b', # 身分証明書番号
r'\b1[3-9]\d{9}\b', # 携帯電話番号
r'[Pp]assword\s*[:=]\s*\S+', # パスワードパターン
]
for pattern in sensitive_patterns:
if re.search(pattern, output):
return False, f"出力に機密情報が含まれている可能性があります: {pattern}"
return True, "通過"サーキットブレーカー機構
Agent の各タスクに明確なリソース上限を設定する：
class CircuitBreaker:
"""Agent サーキットブレーカー：制御不能なループによる巨額の損失を防ぐ"""
def __init__(self,
max_api_calls: int = 10, # 最大 API 呼び出し回数
max_wall_time: int = 300, # 最長実行時間（秒）
max_cost: float = 0.50): # 最大コスト（ドル）
self.max_api_calls = max_api_calls
self.max_wall_time = max_wall_time
self.max_cost = max_cost
self.reset()
def reset(self):
self.api_calls = 0
self.start_time = time.time()
self.total_cost = 0.0
def check(self) -> tuple[bool, str]:
"""サーキットブレークすべきかチェック"""
self.api_calls += 1
elapsed = time.time() - self.start_time
if self.api_calls > self.max_api_calls:
return False, f"API 呼び出し回数が上限を超えました ({self.api_calls}/{self.max_api_calls})"
if elapsed > self.max_wall_time:
return False, f"実行時間が上限を超えました ({elapsed:.0f}s/{self.max_wall_time}s)"
if self.total_cost > self.max_cost:
return False, f"コストが上限を超えました (${self.total_cost:.2f}/${self.max_cost:.2f})"
return True, "正常"Harness Engineering
本番環境の全体ブループリント
Harness Engineering は、Agent システムが開発から本番まで安定稼働するための全般的なエンジニアリング実践である。以下を含む：
可観測性
OpenTelemetry 標準を用いて三種類のデータ収集を確立する：
- Metrics（指標）：Token 消費、遅延分布、エラー率
- Traces（リンクトレーシング）：一回のリクエストが経由する各工程とその所要時間
- Logs（ログ）：各工程の入出力、エラースタック、監査情報
# OpenTelemetry を使用した Agent 呼び出しの計装
from opentelemetry import trace
from opentelemetry.instrumentation.openai import OpenAIInstrumentor
# OpenAI API 呼び出しを自動計装
OpenAIInstrumentor().instrument()
tracer = trace.get_tracer(__name__)
@tracer.start_as_current_span("agent_task")
def run_agent_task(task: str):
# span が自動的に所要時間とコンテキストを記録
with tracer.start_as_current_span("llm_call") as span:
span.set_attribute("task", task)
result = agent.run(task)
span.set_attribute("tokens_used", result.usage.total_tokens)
return resultデプロイ前チェックリスト
Agent を本番環境に投入する前に、以下のチェックを完了する：
☐ SLO 定義
├─ TTFT (初回 Token 遅延) 目標: _____ ms
├─ TPOT (Token あたり生成時間) 目標: _____ ms
└─ 可用性目標: _____%
☐ コスト制御
├─ 単一呼び出し予算上限: $_____
├─ 日次 Token 消費上限: _____ tokens
└─ アラート閾値設定済み
☐ セキュリティ防護
├─ 入力セキュリティチェック有効化済み
├─ 出力コンテンツ審査有効化済み
├─ Agent 行動サーキットブレーカー設定済み
└─ セキュリティ監視アラート設定済み
☐ 災害復旧計画
├─ モデルデグラデーションパス定義済み
├─ クリティカルパスのフォールバックロジック検証済み
└─ 障害復旧訓練完了済み
☐ 評価ベースライン
├─ エンドツーエンド評価スコア: _____
├─ コンポーネント別評価スコア: _____
└─ リグレッションテストパイプライン統合済み
☐ 可観測性
├─ OpenTelemetry 導入済み
├─ 重要指標 Dashboard 構築済み
└─ アラートルール設定・検証済み漸進的リリース戦略
一度に全量リリースしてはならない。漸進的戦略を採用する：
Phase 1: 内部テスト (1-2 週間)
└─ チームメンバーが使用し、初期フィードバックを収集
Phase 2: 小規模カナリア (5% ユーザー)
└─ 新旧システムの評価指標を比較
Phase 3: 範囲拡大 (25% → 50% → 100%)
└─ 各段階で 3-5 日間滞在し指標を観察
Phase 4: 全量リリース
└─ 監視を維持し、継続的最適化サイクルを確立本番運用ベストプラクティス
評価ベースライン管理：
- 現在の本番バージョンをベースラインとし、新バージョンは必ずベースラインを超えなければならない
- 定期的に（毎週）最新データでベースラインと候補バージョンを再テスト
- デプロイパイプラインにベースラインチェックを統合し、基準未達のバージョンは自動遮断
階層的デグラデーション戦略：
- メインモデル不可用 → 代替モデルに切り替え
- 代替モデルも不可用 → キャッシュされた一般的な回答を使用
- キャッシュ未ヒット → 事前設定されたデグラデーション応答テンプレートを返す
コストガバナンス：
- モデル別、ユーザー別、タスクタイプ別の次元で Token 消費を分析
- 異常なコストピークを識別しアラートを設定
- 定期的に審査：不必要な長いコンテキスト、冗長な System Prompt がないか
モデル蒸留：小規模モデルにドメインの専門知識を教える
蒸留が重要な理由
大規模モデル（GPT-4、Claude、Qwen-72B）は優れた品質を提供するが、推論コストと遅延が高い。1分間に数千のリクエストを処理する本番システムでは、トークンコストが莫大になる可能性がある。モデル蒸留は実用的な解決策を提供する：大規模モデルの出力を訓練データとして使用し、小規模モデル（7B-14B）に同じ動作を大幅な低コストで再現できるようにする。
教師モデル (GPT-4, 175B パラメータ)
│
├── ドメインタスクの高品質な応答を生成
│
▼
訓練データ (入力 → 教師出力 ペア)
│
├── 生徒モデルをファインチューニング
│
▼
生徒モデル (7B-14B パラメータ)
│
├── 同じ品質、コスト 10-50 倍低減
└── 遅延 5-10 倍低減蒸留パイプライン
ステップ 1：タスクスコープを定義する
蒸留はタスクが明確に定義され、繰り返し多い場合に最も効果的である。小規模モデルに実行させたい具体的なドメインを特定する：
# 例：カスタマーサポート意図分類
task_definitions = [
{
"name": "intent_classification",
"input_schema": {"user_message": "string", "context": "string"},
"output_schema": {"intent": "string", "confidence": "float", "reasoning": "string"},
},
{
"name": "response_generation",
"input_schema": {"intent": "string", "knowledge": "string", "tone": "string"},
"output_schema": {"response": "string", "sources": ["string"]},
},
]ステップ 2：教師モデルで訓練データを生成する
大規模モデルを使用してドメインの高品質な例を生成する：
from openai import OpenAI
teacher = OpenAI(api_key="...") # GPT-4 または類似モデル
student = OpenAI(base_url="http://localhost:8000/v1") # 小規模モデル
def generate_training_examples(task_def: dict, n_examples: int = 1000) -> list:
examples = []
for i in range(n_examples):
# 多様な入力を生成
input_prompt = f"タスク '{task_def['name']}' の現実的な入力を生成してください。複雑さとエッジケースに変化を持たせてください。"
input_resp = teacher.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": input_prompt}],
)
user_input = input_resp.choices[0].message.content
# 教師の出力を生成
teacher_resp = teacher.chat.completions.create(
model="gpt-4",
messages=[
{"role": "system", "content": f"あなたは {task_def['name']} の専門家です。この出力スキーマに従ってください：{task_def['output_schema']}"},
{"role": "user", "content": user_input},
],
)
teacher_output = teacher_resp.choices[0].message.content
examples.append({"input": user_input, "output": teacher_output})
return examples
training_data = generate_training_examples(task_definitions[0], n_examples=2000)ステップ 3：生徒モデルをファインチューニングする
LoRA またはフルファインチューニングを使用して、教師の出力で小規模モデルを訓練する：
# Hugging Face Transformers + PEFT を使用した LoRA ファインチューニング
from transformers import AutoModelForCausalLM, AutoTokenizer
from peft import LoraConfig, get_peft_model
model_name = "meta-llama/Llama-2-7b-hf"
model = AutoModelForCausalLM.from_pretrained(model_name)
tokenizer = AutoTokenizer.from_pretrained(model_name)
# 効率的なファインチューニングのために LoRA を設定
lora_config = LoraConfig(
r=16, # LoRA ランク
lora_alpha=32,
target_modules=["q_proj", "v_proj"],
lora_dropout=0.05,
bias="none",
task_type="CAUSAL_LM",
)
model = get_peft_model(model, lora_config)
model.print_trainable_parameters()
# 出力：trainable params: 4,194,304 || all params: 6,742,609,920 || trainable%: 0.0622
# 蒸留データで訓練
# ...（標準的な訓練ループ）ステップ 4：評価とイテレーション
ホールドアウトテストセットで生徒モデルの出力と教師の出力を比較する：
def evaluate_student_vs_teacher(test_set: list, teacher_client, student_client) -> dict:
results = {"exact_match": 0, "semantic_similarity": 0, "total": len(test_set)}
for example in test_set:
student_resp = student_client.chat.completions.create(
model="student-7b",
messages=[{"role": "user", "content": example["input"]}],
)
student_output = student_resp.choices[0].message.content
teacher_output = example["output"]
# 完全一致チェック
if student_output.strip() == teacher_output.strip():
results["exact_match"] += 1
# 意味的類似度（embedding を使用）
# ...（出力間のコサイン類似度を計算）
results["exact_match_rate"] = results["exact_match"] / results["total"]
return results
# 目標：教師との意味的類似度 >85%蒸留すべきタイミング
| シナリオ | 推奨 |
|---|---|
| 高頻度・反復タスク（分類、抽出） | 蒸留 — コスト削減が大幅 |
| 創造的・オープンエンドな生成 | 教師モデルを維持 — コストより品質が重要 |
| 遅延に敏感なアプリケーション（リアルタイムチャット） | 蒸留 — 小規模モデルは 5-10 倍高速 |
| まれで複雑な推論タスク | 教師モデルを維持 — 小規模モデルは新しい推論が苦手 |
| ハイブリッド：単純タスク + 複雑なエッジケース | ルーティング — 小規模モデルで 80% を処理し、20% を教師にエスカレーション |
本番ベストプラクティス：モニタリング、カナリアリリース、A/B テスト
可観測性スタック
本番環境の Agent システムには包括的な可観測性が必要である。以下の監視レイヤーを構築する：
1. アプリケーションレベルの指標
import time
from prometheus_client import Counter, Histogram, Gauge
# リクエスト指標
request_counter = Counter('agent_requests_total', '総リクエスト数', ['model', 'intent', 'status'])
request_latency = Histogram('agent_request_duration_seconds', 'リクエスト遅延', ['model', 'intent'])
active_sessions = Gauge('agent_active_sessions', 'アクティブなセッション数')
# Token 使用量指標
token_usage = Counter('agent_tokens_total', 'Token 使用量', ['model', 'type']) # type: input/output
# 品質指標
hallucination_rate = Gauge('agent_hallucination_rate', '推定ハルシネーション率')
user_satisfaction = Histogram('agent_user_satisfaction', 'ユーザー満足度スコア', buckets=[1, 2, 3, 4, 5])
def track_request(model: str, intent: str, status: str, latency: float, tokens_in: int, tokens_out: int):
request_counter.labels(model=model, intent=intent, status=status).inc()
request_latency.labels(model=model, intent=intent).observe(latency)
token_usage.labels(model=model, type='input').inc(tokens_in)
token_usage.labels(model=model, type='output').inc(tokens_out)2. 分散トレーシング
OpenTelemetry を使用して Agent パイプライン全体でリクエストをトレースする：
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
trace.set_tracer_provider(TracerProvider())
tracer = trace.get_tracer(__name__)
def process_user_request(user_input: str):
with tracer.start_as_current_span("process_request") as span:
span.set_attribute("user.input_length", len(user_input))
with tracer.start_as_current_span("retrieve_context") as retrieve_span:
context = retrieve_relevant_docs(user_input)
retrieve_span.set_attribute("docs.retrieved", len(context))
with tracer.start_as_current_span("llm_generate") as llm_span:
llm_span.set_attribute("model", "gpt-4")
response = call_llm(user_input, context)
llm_span.set_attribute("tokens.used", response.usage.total_tokens)
return response3. ロギング戦略
import structlog
logger = structlog.get_logger()
def log_agent_event(event_type: str, **kwargs):
logger.info(
event_type,
session_id=kwargs.get("session_id"),
user_id=kwargs.get("user_id"),
model=kwargs.get("model"),
intent=kwargs.get("intent"),
latency_ms=kwargs.get("latency_ms"),
tokens_in=kwargs.get("tokens_in"),
tokens_out=kwargs.get("tokens_out"),
error=kwargs.get("error"),
)
# 使用例
log_agent_event(
"request_completed",
session_id="abc123",
user_id="user_456",
model="gpt-4",
intent="faq_answer",
latency_ms=1234,
tokens_in=500,
tokens_out=200,
)カナリアリリース戦略
リスクを最小化するために変更を段階的にロールアウトする：
Phase 1: カナリア（トラフィックの 5%）
└─ 小規模サブセットで新モデル/設定を実行
└─ エラー率、遅延、満足度を監視
└─ 期間：1-3 日
Phase 2: 拡大（25% → 50%）
└─ トラフィックを徐々に増加
└─ ベースラインと指標を比較
└─ 各段階の期間：3-5 日
Phase 3: 全量ロールアウト（100%）
└─ 全トラフィックを新バージョンに
└─ 監視を継続
└─ クイックロールバックのために旧バージョンを保持# ユーザー ID ハッシュに基づくシンプルなカナリアルーティング
import hashlib
def route_to_version(user_id: str, canary_percent: int = 5) -> str:
hash_val = int(hashlib.md5(user_id.encode()).hexdigest(), 16)
bucket = hash_val % 100
return "canary" if bucket < canary_percent else "stable"
# リクエストハンドラ内
def handle_request(user_id: str, user_input: str):
version = route_to_version(user_id, canary_percent=5)
if version == "canary":
response = call_new_model(user_input)
else:
response = call_stable_model(user_input)
# バージョンごとに指標を追跡
track_request(model=version, ...)
return responseA/B テストフレームワーク
2つのバージョンを直接比較して影響を測定する：
from dataclasses import dataclass
from enum import Enum
class Variant(Enum):
CONTROL = "control"
TREATMENT = "treatment"
@dataclass
class ABTestResult:
variant: Variant
total_requests: int
avg_latency_ms: float
success_rate: float
user_satisfaction: float # 1-5 スケール
def run_ab_test(user_id: str, user_input: str) -> tuple[str, str]:
"""(バリアント, 応答) を返す"""
# ユーザー ID に基づく一貫した割り当て
hash_val = int(hashlib.md5(user_id.encode()).hexdigest(), 16)
variant = Variant.CONTROL if hash_val % 2 == 0 else Variant.TREATMENT
if variant == Variant.CONTROL:
response = call_control_model(user_input)
else:
response = call_treatment_model(user_input)
return variant.value, response
def analyze_ab_results(results_a: list, results_b: list) -> dict:
"""コントロールとトリートメントの指標を比較"""
import statistics
def compute_metrics(results):
return {
"count": len(results),
"avg_latency": statistics.mean([r["latency_ms"] for r in results]),
"success_rate": sum(1 for r in results if r["success"]) / len(results),
}
metrics_a = compute_metrics(results_a)
metrics_b = compute_metrics(results_b)
return {
"control": metrics_a,
"treatment": metrics_b,
"latency_improvement": (metrics_a["avg_latency"] - metrics_b["avg_latency"]) / metrics_a["avg_latency"] * 100,
"success_rate_delta": (metrics_b["success_rate"] - metrics_a["success_rate"]) * 100,
}アラートルール
重要な条件に対してアラートを設定する：
# Prometheus アラートルール
groups:
- name: agent_alerts
rules:
- alert: HighErrorRate
expr: rate(agent_requests_total{status="error"}[5m]) / rate(agent_requests_total[5m]) > 0.05
for: 5m
labels:
severity: critical
annotations:
summary: "Agent エラー率が 5% を超えています"
- alert: HighLatency
expr: histogram_quantile(0.95, rate(agent_request_duration_seconds_bucket[5m])) > 10
for: 5m
labels:
severity: warning
annotations:
summary: "P95 遅延が 10 秒を超えています"
- alert: TokenCostSpike
expr: rate(agent_tokens_total[1h]) > 100000
for: 30m
labels:
severity: warning
annotations:
summary: "Token 使用量のスパイクを検出"RIDE 手法：AI 駆動のビジネスインパクトのためのフレームワーク
課題
AI の導入を急ぐ組織は、しばしば以下の二つの罠のいずれかに陥る：
- 問題を探している解決策：実際のビジネスニーズに対応していない印象的な AI デモを構築する
- 分析麻痺：何も出荷せずに AI ツールの無限の評価を続ける
RIDE 手法は、実際のビジネス価値を提供する AI 施策を選択、実装、測定するための構造化されたアプローチを提供する。
RIDE: Research → Implement → Deliver → Enhance
┌─────────────────────────────────────────────────────────────┐
│ RIDE サイクル │
│ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────┐ │
│ │ Research │───▶│Implement │───▶│ Deliver │───▶│Enhance│ │
│ │ │ │ │ │ │ │ │ │
│ └──────────┘ └──────────┘ └──────────┘ └──────┘ │
│ ▲ │ │
│ └──────────────────────────────────────────────┘ │
│ （イテレーション） │
└─────────────────────────────────────────────────────────────┘フェーズ 1: Research — 正しい問題を選択する
目標：ビジネス優先事項に合致する高インパクトで実現可能な AI ユースケースを特定する。
主要な活動：
- ビジネスの課題をマッピング：ステークホルダーへのインタビュー、サポートチケットの分析、プロセスのボトルネックの review
- Impact-Feasibility マトリックスで機会をスコアリング：
@dataclass
class UseCase:
name: str
description: str
business_impact: int # 1-10：収益への影響、コスト削減、顧客満足度
technical_feasibility: int # 1-10：データの可用性、モデルの能力、統合の複雑さ
time_to_value: int # MVP までの週数
stakeholders: list[str]
def score_use_case(uc: UseCase) -> float:
"""スコアが高い = AI 施策の候補としてより優れている"""
return (uc.business_impact * 0.4 + uc.technical_feasibility * 0.3 +
(10 - uc.time_to_value / 4) * 0.3) # 時間を 1-10 スケールに正規化
# スコアリング例
use_cases = [
UseCase("FAQ Bot", "カスタマー FAQ 応答を自動化", 7, 9, 4, ["サポート", "エンジニアリング"]),
UseCase("Code Review", "AI 支援コードレビュー", 6, 7, 8, ["エンジニアリング"]),
UseCase("Demand Forecast", "製品需要を予測", 9, 5, 12, ["プロダクト", "サプライチェーン"]),
]
for uc in sorted(use_cases, key=score_use_case, reverse=True):
print(f"{uc.name}: score={score_use_case(uc):.1f}, impact={uc.business_impact}, feasibility={uc.technical_feasibility}")
# 出力：
# FAQ Bot: score=8.2, impact=7, feasibility=9
# Code Review: score=6.7, impact=6, feasibility=7
# Demand Forecast: score=6.1, impact=9, feasibility=5- ステークホルダーとともに検証：上位候補を提示し、承認を得て、成功基準を定義する
成果物：明確な成功指標を持つ 2-3 の優先ユースケースのリスト。
フェーズ 2: Implement — MVP を構築する
目標：コア機能に注力して、動作するプロトタイプを素早く出荷する。
主要原則：
- 最もシンプルなアプローチから始める：ルールベース → RAG → ファインチューニングモデル（必要に応じてのみエスカレーション）
- 既存のツールを使用する：必要でない限りインフラを自作しない
- 初日から測定する：前セクションの可観測性スタックで MVP を計装する
実装チェックリスト：
1-2 週目：基盤
□ 開発環境のセットアップ
□ データソースとアクセスの定義
□ 基本的なプロンプトテンプレートの作成
□ 評価データセットの作成（50-100 例）
3-4 週目：コア機能
□ 検索パイプラインの実装（RAG の場合）
□ Agent ループの構築（ツール使用の場合）
□ 既存システムとの統合（API、データベース）
□ 基本的なエラーハンドリングとフォールバックの追加
5-6 週目：品質とテスト
□ 評価スイートの実行、プロンプトのイテレーション
□ 安全ガードレールの追加（コンテンツフィルタリング、PII 検出）
□ 5-10 人の内部ユーザーでのユーザーテスト
□ 重要な問題の修正
7-8 週目：デプロイ準備
□ モニタリングとアラートの設定
□ 一般的な問題のランブックを作成
□ ロールバック計画の準備
□ ステージングにデプロイ、負荷テストを実行フェーズ 3: Deliver — ビジネスインパクトを測定する
目標：Research で定義した成功基準に対する実際の影響を定量化する。
ユースケースタイプ別の主要指標：
| ユースケースタイプ | 主要指標 | 副次指標 |
|---|---|---|
| カスタマーサポートボット | チケット防御率、解決時間 | 顧客満足度（CSAT）、チケットあたりのコスト |
| コードレビューアシスタント | レビュー所要時間、欠陥逸脱率 | 開発者満足度、コード品質スコア |
| コンテンツ生成 | コンテンツ制作時間、エンゲージメント指標 | ブランド一貫性スコア、編集レビュー合格率 |
| データ分析 Agent | 分析所要時間、インサイトの品質 | ステークホルダー満足度、意思決定速度 |
インパクト計算例：
# FAQ Bot の前後比較
before = {
"monthly_tickets": 5000,
"avg_resolution_time_hours": 24,
"cost_per_ticket": 15, # 人間のエージェントコスト
"csat_score": 3.2,
}
after = {
"monthly_tickets": 5000,
"bot_deflection_rate": 0.65, # 65% がボットで処理
"avg_resolution_time_hours": 0.5, # ボット処理の場合
"cost_per_ticket_bot": 0.50, # API コスト
"cost_per_ticket_human": 15, # エスカレーションの場合
"csat_score": 4.1,
}
# インパクトを計算
bot_handled = after["monthly_tickets"] * after["bot_deflection_rate"]
human_handled = after["monthly_tickets"] - bot_handled
monthly_cost_before = before["monthly_tickets"] * before["cost_per_ticket"]
monthly_cost_after = (bot_handled * after["cost_per_ticket_bot"] +
human_handled * after["cost_per_ticket_human"])
monthly_savings = monthly_cost_before - monthly_cost_after
annual_savings = monthly_savings * 12
print(f"変更前の月間コスト: ${monthly_cost_before:,.0f}")
print(f"変更後の月間コスト: ${monthly_cost_after:,.0f}")
print(f"年間削減額: ${annual_savings:,.0f}")
print(f"CSAT 改善: {after['csat_score'] - before['csat_score']:.1f} ポイント")
# 出力：
# 変更前の月間コスト: $75,000
# 変更後の月間コスト: $29,875
# 年間削減額: $541,500
# CSAT 改善: 0.9 ポイントフェーズ 4: Enhance — イテレーションと拡張
目標：データとフィードバックに基づいてシステムを継続的に改善する。
強化戦略：
- プロンプト最適化：評価データを使用してプロンプトを改善（評価チャプターを参照）
- RAG の改善：ドキュメントの追加、チャンキングの改善、リランキングの追加
- モデルアップグレード：コスト削減のために小規模モデルに蒸留（蒸留セクションを参照）
- 機能拡張：ユーザーフィードバックに基づいて新機能を追加
- プロセス統合：既存のワークフローとの統合を深化
イテレーションのペース：
毎週：
- エラーログとユーザーフィードバックをレビュー
- 新しい例で評価データセットを更新
- 重要なバグを修正
毎月：
- 完全な評価スイートを実行
- 指標のトレンドを分析
- 次のイテレーションを計画
四半期ごと：
- 目標に対するビジネスインパクトをレビュー
- アップグレードのためのモデル/プロバイダー状況を評価
- 戦略的な強化を計画RIDE の実践：よくある落とし穴
| 落とし穴 | 回避方法 |
|---|---|
| Research をスキップして実装に飛び込む | 常にステークホルダーインタビューとユースケーススコアリングから始める |
| 配信前に何ヶ月も構築する | 8 週間の MVP 締切を設定；測定可能なものを出荷する |
| 技術指標（遅延、精度）のみを測定する | 事前にビジネス指標を定義；コスト削減、時間節約を追跡 |
| AI を一度限りのプロジェクトとして扱う | 継続的なイテレーションを計画；AI システムには継続的なメンテナンスが必要 |
| 安全とコンプライアンスを無視する | 初日からガードレールを構築；後から付け加えない |
付録：重要概念クイックリファレンス
| 概念 | 一言解説 |
|---|---|
| Tokenization | テキストをモデルが処理可能な数値 ID 系列に変換する |
| Embedding | 離散 ID を意味情報を含む密なベクトルに写像する |
| Attention | モデルが各 token を処理する際に系列内のすべての関連位置に注目できるようにする |
| RAG | まず関連知識を検索し、次にモデルが知識に基づいて回答を生成する |
| ReAct | モデルに思考（Reasoning）と行動（Action）を交互に循環させる |
| Function Calling | モデルが純粋なテキスト応答ではなく構造化されたツール呼び出し命令を出力する |
| MCP | Anthropic が提唱するツール標準化プロトコル。ツール定義と使用を疎結合にする |
| Plan & Execute | まず完全な行動計画を策定し、審査通過後に段階的に実行する |
| HyDE | まず仮想回答を生成し、元の質問ではなく仮想回答を用いて検索する |
| Lost in the Middle | モデルが長いコンテキストの中間部分の情報処理能力を著しく低下させる現象 |
| Mixture-of-Agents | 複数の異なるモデルが同一タスクを処理し、集約器が最適結果を統合する |
| LoRA | 低ランク行列アダプタを学習することで効率的なファインチューニングを実現する |
| 蒸留 | 大規模モデルの出力を学習データとして小規模モデルを教える |
| Skill | 専門知識を再利用可能なモジュール化機能単位にカプセル化する |
| LLM-as-Judge | 大規模モデルを評価員として出力の自動採点に使用する |
| SLO | サービスレベル目標。TTFT、TPOT、可用性など |
本ガイドは Agent エンジニアリングのベストプラクティスに基づいて執筆され、LLM の基礎から本番運用までの完全なパスをカバーしている。技術分野は日進月歩であり、コミュニティの発展を継続的に注視し、本ガイドの方法論を最新のツールと組み合わせて活用することを推奨する。
総合演習プロジェクト：インテリジェント Q&A Agent の構築
プロジェクトの物語：Part 1-5 では、Agent エンジニアリングの各構成要素を個別に学んできました。今度はそれらをすべて組み合わせます。新入社員向け Q&A Agent を構築します。シンプルな API コールから始まり、RAG、ツール呼び出し、メモリ、スキル、評価、デプロイを備えた本番グレードの Agent へと進化していくシステムです。各ステージは実際のエンジニアリングのマイルストーンに対応しています。
6.1 環境構築と基本的な会話
スタート地点
あなたの会社には問題があります：新入社員がオンボーディング、福利厚生、ツール、プロセスについて同じ質問を繰り返し聞いてきます。HR は同じ回答を繰り返すのに何時間も費やしています。あなたの目標は、これらの質問に正確に答えられる AI Agent を構築することです。
ステージ 1 の目標：基本的な LLM 対話を動作させる。
プロジェクトのセットアップ
# プロジェクトディレクトリを作成
mkdir qa-agent && cd qa-agent
# Python 環境をセットアップ
python -m venv .venv
source .venv/bin/activate
# 依存パッケージをインストール
pip install openai python-dotenv# .env
OPENAI_API_KEY=sk-your-key-here最初の会話
# chat.py
import os
from openai import OpenAI
from dotenv import load_dotenv
load_dotenv()
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
def chat(user_message: str) -> str:
response = client.chat.completions.create(
model="gpt-4",
messages=[
{"role": "system", "content": "You are a helpful onboarding assistant for new employees."},
{"role": "user", "content": user_message},
],
)
return response.choices[0].message.content
# テスト
print(chat("What's the dress code?"))
# 出力: "Our dress code is business casual..."マルチターン会話
単発のチャットでは不十分です。社員はフォローアップの質問をします。会話履歴を追加しましょう：
# multi_turn_chat.py
conversation_history = [
{"role": "system", "content": "You are a helpful onboarding assistant for new employees at Acme Corp."}
]
def chat_with_history(user_message: str) -> str:
conversation_history.append({"role": "user", "content": user_message})
response = client.chat.completions.create(
model="gpt-4",
messages=conversation_history,
)
assistant_reply = response.choices[0].message.content
conversation_history.append({"role": "assistant", "content": assistant_reply})
return assistant_reply
# マルチターンのテスト
print(chat_with_history("What's the dress code?"))
print(chat_with_history("What about on Fridays?")) # フォローアップ質問
print(chat_with_history("And for client meetings?")) # さらなるフォローアップトークン予算の管理
# token_tracker.py
import tiktoken
def count_tokens(messages: list, model: str = "gpt-4") -> int:
encoding = tiktoken.encoding_for_model(model)
total = 0
for msg in messages:
total += len(encoding.encode(msg["content"])) + 4 # メッセージごとのオーバーヘッド
return total
def chat_within_budget(user_message: str, max_tokens: int = 4000) -> str:
conversation_history.append({"role": "user", "content": user_message})
# 予算超過の場合、履歴をトリミング
while count_tokens(conversation_history) > max_tokens:
# 最も古い非 system メッセージペアを削除
non_system = [m for m in conversation_history if m["role"] != "system"]
if len(non_system) >= 2:
conversation_history.remove(non_system[0])
conversation_history.remove(non_system[1])
else:
break
response = client.chat.completions.create(
model="gpt-4",
messages=conversation_history,
)
reply = response.choices[0].message.content
conversation_history.append({"role": "assistant", "content": reply})
return replyステージ 1 完了：基本的なマルチターンチャットボットができました。しかし、LLM が学習した内容しか知らず、会社の固有ポリシーについては何も知りません。
6.2 RAG：企業知識への接続
問題
print(chat_with_history("What's the parental leave policy?"))
# 出力: "I don't have specific information about Acme Corp's parental leave policy..."LLM は会社の内部ドキュメントを知りません。Retrieval-Augmented Generation（RAG） が必要です。
ステップ 1：ナレッジベースの準備
# knowledge_base.py
import os
from pathlib import Path
# サンプルの会社ドキュメント（実際にはドキュメントストアから読み込み）
documents = [
{
"id": "doc_001",
"title": "Employee Handbook - Leave Policies",
"content": """Acme Corp provides the following leave benefits:
- Annual Leave: 20 days per year, prorated for partial years
- Sick Leave: 10 days per year
- Parental Leave: 16 weeks paid leave for primary caregivers, 8 weeks for secondary caregivers
- Bereavement Leave: 5 days for immediate family members
All leave requests must be submitted through the HR portal at least 2 weeks in advance, except for sick leave which can be reported same-day."""
},
{
"id": "doc_002",
"title": "Employee Handbook - Dress Code",
"content": """Acme Corp Dress Code:
- Regular days: Business casual (collared shirts, slacks, closed-toe shoes)
- Casual Fridays: Jeans and casual wear allowed, but no flip-flops or gym clothes
- Client meetings: Business formal (suit and tie for men, business suit or dress for women)
- Remote work days: No dress code, but camera-on for meetings
When in doubt, err on the side of being more formal."""
},
# ... その他のドキュメント
]ステップ 2：検索パイプラインの構築
# rag_pipeline.py
from openai import OpenAI
import numpy as np
client = OpenAI()
def get_embedding(text: str) -> list[float]:
response = client.embeddings.create(
model="text-embedding-3-small",
input=text,
)
return response.data[0].embedding
def build_vector_store(docs: list) -> dict:
"""全ドキュメントを埋め込み、シンプルなベクトルインデックスとして保存"""
vector_store = {}
for doc in docs:
embedding = get_embedding(doc["content"])
vector_store[doc["id"]] = {
"embedding": embedding,
"content": doc["content"],
"title": doc["title"],
}
return vector_store
def search(query: str, vector_store: dict, top_k: int = 3) -> list[dict]:
"""最も関連性の高い top-k ドキュメントを取得"""
query_embedding = get_embedding(query)
results = []
for doc_id, doc_data in vector_store.items():
similarity = cosine_similarity(query_embedding, doc_data["embedding"])
results.append({
"doc_id": doc_id,
"title": doc_data["title"],
"content": doc_data["content"],
"score": similarity,
})
results.sort(key=lambda x: x["score"], reverse=True)
return results[:top_k]
def cosine_similarity(a: list, b: list) -> float:
a_arr = np.array(a)
b_arr = np.array(b)
return np.dot(a_arr, b_arr) / (np.linalg.norm(a_arr) * np.linalg.norm(b_arr))ステップ 3：取得したコンテキストで生成を強化
# rag_chat.py
def rag_chat(user_message: str, vector_store: dict) -> str:
# ステップ 1：関連ドキュメントを検索
relevant_docs = search(user_message, vector_store, top_k=3)
# ステップ 2：取得したドキュメントからコンテキストを構築
context = "\n\n".join([
f"[Source: {doc['title']}]\n{doc['content']}"
for doc in relevant_docs
])
# ステップ 3：コンテキスト付きで回答を生成
system_prompt = f"""You are a helpful onboarding assistant for Acme Corp.
Use the following company documents to answer questions. If the answer is not in the documents, say so honestly.
Always cite which document you're referencing.
--- Company Documents ---
{context}
--- End Documents ---"""
response = client.chat.completions.create(
model="gpt-4",
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_message},
],
)
return response.choices[0].message.content
# テスト
vector_store = build_vector_store(documents)
print(rag_chat("What's the parental leave policy?", vector_store))
# 出力: "According to the Employee Handbook - Leave Policies, Acme Corp provides:
# - 16 weeks paid leave for primary caregivers
# - 8 weeks for secondary caregivers
# All requests must be submitted through the HR portal at least 2 weeks in advance."ステージ 2 完了：ボットが会社ドキュメントから回答できるようになりました。しかし、話す ことしかできず、休暇申請の提出やカレンダーの空き状況確認などの行動 はできません。
6.3 Agent のツール呼び出しとプランニング
問題
社員がこう尋ねました：「来月の月曜日から水曜日までの休暇申請を出してもらえますか？」
ボットはポリシーを説明できますが、実際に申請を提出することはできません。ツール呼び出し が必要です。
ステップ 1：ツールの定義
# tools.py
import json
tools = [
{
"type": "function",
"function": {
"name": "submit_leave_request",
"description": "Submit a leave request to the HR system",
"parameters": {
"type": "object",
"properties": {
"leave_type": {
"type": "string",
"enum": ["annual", "sick", "parental", "bereavement"],
"description": "Type of leave"
},
"start_date": {"type": "string", "description": "Start date (YYYY-MM-DD)"},
"end_date": {"type": "string", "description": "End date (YYYY-MM-DD)"},
"reason": {"type": "string", "description": "Reason for leave"},
},
"required": ["leave_type", "start_date", "end_date"],
},
},
},
{
"type": "function",
"function": {
"name": "check_leave_balance",
"description": "Check remaining leave balance for an employee",
"parameters": {
"type": "object",
"properties": {
"employee_id": {"type": "string", "description": "Employee ID"},
"leave_type": {"type": "string", "description": "Type of leave to check"},
},
"required": ["employee_id"],
},
},
},
{
"type": "function",
"function": {
"name": "search_knowledge_base",
"description": "Search the company knowledge base for policies and procedures",
"parameters": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "Search query"},
},
"required": ["query"],
},
},
},
]
# モック実装
def submit_leave_request(leave_type: str, start_date: str, end_date: str, reason: str = "") -> dict:
# 本番環境では、HR API を呼び出す
return {"status": "submitted", "request_id": "LR-2024-001", "leave_type": leave_type, "dates": f"{start_date} to {end_date}"}
def check_leave_balance(employee_id: str, leave_type: str = None) -> dict:
# モックデータ
balances = {"annual": 15, "sick": 8, "parental": 0, "bereavement": 5}
if leave_type:
return {"employee_id": employee_id, "leave_type": leave_type, "remaining_days": balances.get(leave_type, 0)}
return {"employee_id": employee_id, "balances": balances}
def search_knowledge_base(query: str) -> str:
# 前のセクションの RAG 検索を再利用
results = search(query, vector_store, top_k=2)
return "\n\n".join([f"[{r['title']}]: {r['content'][:200]}..." for r in results])
TOOL_IMPLEMENTATIONS = {
"submit_leave_request": submit_leave_request,
"check_leave_balance": check_leave_balance,
"search_knowledge_base": search_knowledge_base,
}ステップ 2：ReAct ループの実装
# agent.py
def agent_chat(user_message: str, max_iterations: int = 5) -> str:
messages = [
{"role": "system", "content": """You are an onboarding assistant for Acme Corp.
You can use tools to help answer questions and perform actions.
Always think step by step. If you need information, use search_knowledge_base.
If the user wants to perform an action, use the appropriate tool.
After getting tool results, provide a clear summary to the user."""},
{"role": "user", "content": user_message},
]
for i in range(max_iterations):
response = client.chat.completions.create(
model="gpt-4",
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)
print(f" [Tool Call] {func_name}({func_args})")
# ツールを実行
result = TOOL_IMPLEMENTATIONS[func_name](**func_args)
# ツール結果をメッセージに追加
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": json.dumps(result),
})
return "I wasn't able to complete the task within the allowed steps."
# テスト
print(agent_chat("How many annual leave days do I have left? My employee ID is EMP-042."))
# 出力:
# [Tool Call] check_leave_balance({'employee_id': 'EMP-042', 'leave_type': 'annual'})
# "You have 15 annual leave days remaining."
print(agent_chat("Can you submit annual leave for me from Dec 23 to Dec 27?"))
# 出力:
# [Tool Call] submit_leave_request({'leave_type': 'annual', 'start_date': '2024-12-23', 'end_date': '2024-12-27'})
# "Your leave request has been submitted! Request ID: LR-2024-001, covering Dec 23-27, 2024."ステップ 3：複雑なタスクへのプランニング追加
マルチステップのタスクでは、Agent は実行前にプランニングする必要があります：
# planner.py
def plan_and_execute(user_request: str) -> str:
# ステップ 1：プランを生成
plan_response = client.chat.completions.create(
model="gpt-4",
messages=[
{"role": "system", "content": """You are a task planner. Break down the user's request into concrete steps.
For each step, specify which tool to use and what arguments to pass.
Output a JSON array of steps."""},
{"role": "user", "content": user_request},
],
)
plan_text = plan_response.choices[0].message.content
print(f"Plan: {plan_text}")
# ステップ 2：Agent を使って各ステップを実行
# （本番環境では、プランをパースして検証付きでステップごとに実行する）
return agent_chat(user_request)ステージ 3 完了：ボットがツールを呼び出してアクションを実行できるようになりました。しかし、毎回ゼロから会話が始まり、過去のやり取りを覚えていません。
6.4 メモリとスキル：Agent を時間とともに賢くする
問題
一人の社員が 3 回の別々の会話をしました：
- 「来月の月曜日から入社しますが、何を持っていけばいいですか？」
- 「ありがとうございます。ちなみに、社員番号は EMP-042 です。」
- 「休暇残を確認してもらえますか？」
Agent は社員番号を全く知りません。各会話が孤立しています。
ステップ 1：短期メモリ（会話バッファ）
# memory.py
from dataclasses import dataclass, field
@dataclass
class ConversationBuffer:
max_tokens: int = 4000
messages: list = field(default_factory=list)
summary: str = ""
def add_message(self, role: str, content: str):
self.messages.append({"role": role, "content": content})
self._trim_if_needed()
def _trim_if_needed(self):
token_count = count_tokens(self.messages)
if token_count > self.max_tokens:
# 古いメッセージを要約し、最近のメッセージを保持
old_messages = self.messages[:len(self.messages)//2]
self.summary = self._summarize(old_messages)
self.messages = self.messages[len(self.messages)//2:]
def _summarize(self, messages: list) -> str:
response = client.chat.completions.create(
model="gpt-4",
messages=[
{"role": "system", "content": "Summarize this conversation in 2-3 sentences, focusing on key facts and decisions."},
*messages,
],
)
return response.choices[0].message.content
def get_context(self) -> list:
context = []
if self.summary:
context.append({"role": "system", "content": f"Previous conversation summary: {self.summary}"})
context.extend(self.messages)
return contextステップ 2：長期メモリ（ユーザープロファイルストア）
# long_term_memory.py
import json
from pathlib import Path
USER_PROFILES_DIR = Path("user_profiles")
USER_PROFILES_DIR.mkdir(exist_ok=True)
def save_user_fact(user_id: str, fact: str):
"""ユーザーについての事実を長期プロファイルに保存"""
profile_path = USER_PROFILES_DIR / f"{user_id}.json"
profile = {}
if profile_path.exists():
profile = json.loads(profile_path.read_text())
if "facts" not in profile:
profile["facts"] = []
profile["facts"].append(fact)
profile_path.write_text(json.dumps(profile, indent=2))
def get_user_facts(user_id: str) -> list[str]:
"""ユーザーについて既知の全事実を取得"""
profile_path = USER_PROFILES_DIR / f"{user_id}.json"
if not profile_path.exists():
return []
profile = json.loads(profile_path.read_text())
return profile.get("facts", [])
def extract_and_save_facts(user_id: str, messages: list):
"""LLM を使って会話から重要な事実を抽出し保存"""
response = client.chat.completions.create(
model="gpt-4",
messages=[
{"role": "system", "content": """Extract important facts about the user from this conversation.
Focus on: name, employee ID, department, preferences, upcoming events, action items.
Output a JSON array of fact strings. If no important facts, output []."""},
*messages,
],
)
facts = json.loads(response.choices[0].message.content)
for fact in facts:
save_user_fact(user_id, fact)ステップ 3：スキル——再利用可能なワークフロー
# skills/onboarding_guide.md
"""
---
name: onboarding_guide
description: Guide new employees through their first week
triggers: new employee, first day, onboarding, getting started
---
# Onboarding Guide Skill
## Day 1 Checklist
1. Verify IT setup (laptop, accounts, VPN)
2. Introduce to team via Slack
3. Share key documents: handbook, org chart, tools guide
4. Schedule 1:1 with manager for week overview
## Week 1 Priorities
- Complete mandatory training modules (compliance, security)
- Set up development environment (if engineer)
- Attend team standup meetings
- Read team's project documentation
## Common First-Week Questions
- "How do I submit expenses?" → Use Concur, submit within 30 days
- "What's the wifi password?" → Provided on IT setup sheet
- "Who do I talk about benefits?" → HR portal or email [email protected]
"""
# skill_loader.py
from pathlib import Path
def load_skill(skill_name: str) -> str:
skill_path = Path(f"skills/{skill_name}.md")
if not skill_path.exists():
return ""
return skill_path.read_text()
def find_relevant_skill(query: str, available_skills: list[str]) -> str | None:
"""クエリに基づいてアクティブ化するスキルを判定"""
skills_info = []
for skill_name in available_skills:
content = load_skill(skill_name)
# フロントマターから説明を抽出
skills_info.append(f"- {skill_name}: {content[:200]}")
response = client.chat.completions.create(
model="gpt-4",
messages=[
{"role": "system", "content": "Given the user query, which skill should be activated? Output just the skill name, or 'none' if no skill is relevant."},
{"role": "user", "content": f"Query: {query}\n\nAvailable skills:\n" + "\n".join(skills_info)},
],
)
result = response.choices[0].message.content.strip()
return result if result != "none" else None統合：メモリ対応 Agent
# memory_agent.py
def memory_aware_agent_chat(user_id: str, user_message: str) -> str:
# ユーザーの長期ファクトを読み込み
user_facts = get_user_facts(user_id)
facts_context = "\n".join(user_facts) if user_facts else "No previous facts known."
# スキルをアクティブ化するべきかチェック
skill_name = find_relevant_skill(user_message, ["onboarding_guide"])
skill_context = load_skill(skill_name) if skill_name else ""
system_prompt = f"""You are an onboarding assistant for Acme Corp.
Known facts about this user:
{facts_context}
{f'Active skill: {skill_context}' if skill_context else ''}
Use the user's known facts to personalize responses.
If the user shares new important information, note it for future reference."""
messages = [{"role": "system", "content": system_prompt}]
messages.extend(conversation_buffer.get_context())
messages.append({"role": "user", "content": user_message})
response = client.chat.completions.create(
model="gpt-4",
messages=messages,
tools=tools,
tool_choice="auto",
)
reply = response.choices[0].message.content
# メモリを更新
conversation_buffer.add_message("user", user_message)
conversation_buffer.add_message("assistant", reply)
extract_and_save_facts(user_id, messages)
return replyステージ 4 完了：Agent がユーザーを記憶し、スキルをアクティブ化し、パーソナライズされた回答を提供できるようになりました。しかし、本当に良い回答を出しているかどうか、どうやって確認すればよいでしょうか？
6.5 評価と反復最適化
問題
多くのものを構築しましたが、それがどれだけうまく機能しているか分かりません。正しい回答を出しているか？ハルシネーションを起こしていないか？重要なコンテキストを見落としていないか？
ステップ 1：評価データセットの構築
# eval_dataset.py
eval_cases = [
{
"id": "eval_001",
"input": "What's the parental leave policy?",
"expected_output": "16 weeks for primary caregivers, 8 weeks for secondary caregivers",
"required_sources": ["Employee Handbook - Leave Policies"],
"category": "factual_recall",
},
{
"id": "eval_002",
"input": "How do I submit a leave request?",
"expected_output": "Through the HR portal, at least 2 weeks in advance",
"required_sources": ["Employee Handbook - Leave Policies"],
"category": "procedural",
},
{
"id": "eval_003",
"input": "What should I wear to a client meeting?",
"expected_output": "Business formal: suit and tie for men, business suit or dress for women",
"required_sources": ["Employee Handbook - Dress Code"],
"category": "factual_recall",
},
{
"id": "eval_004",
"input": "Can you submit a sick leave request for me today?",
"expected_behavior": "Should call submit_leave_request tool with leave_type='sick'",
"category": "tool_use",
},
{
"id": "eval_005",
"input": "What's the meaning of life?",
"expected_behavior": "Should politely decline or redirect to onboarding topics",
"category": "boundary",
},
]ステップ 2：自動評価
# evaluator.py
def evaluate_factual_recall(agent_fn, case: dict) -> dict:
"""Agent がドキュメントから事実を正しく思い出せたか評価"""
response = agent_fn(case["input"])
# チェック 1：回答に期待情報が含まれているか？
expected_keywords = case["expected_output"].lower().split()
response_lower = response.lower()
keyword_hits = sum(1 for kw in expected_keywords if kw in response_lower)
recall_score = keyword_hits / len(expected_keywords)
# チェック 2：正しいソースを引用したか？
source_cited = any(src.lower() in response.lower() for src in case["required_sources"])
return {
"case_id": case["id"],
"category": case["category"],
"recall_score": recall_score,
"source_cited": source_cited,
"passed": recall_score > 0.7 and source_cited,
"response": response[:200],
}
def evaluate_tool_use(agent_fn, case: dict) -> dict:
"""Agent がツールを正しく使ったか評価"""
# 実行中のツール呼び出しをキャプチャ
tool_calls_made = []
original_implementations = {}
# ツールをラップして呼び出しをキャプチャ
for name, impl in TOOL_IMPLEMENTATIONS.items():
original_implementations[name] = impl
def make_wrapper(n):
def wrapper(*args, **kwargs):
tool_calls_made.append({"name": n, "args": kwargs})
return original_implementations[n](*args, **kwargs)
return wrapper
TOOL_IMPLEMENTATIONS[name] = make_wrapper(name)
try:
response = agent_fn(case["input"])
expected_tool = case["expected_behavior"].split("'")[1] if "'" in case["expected_behavior"] else ""
correct_tool_called = any(tc["name"] == expected_tool for tc in tool_calls_made)
return {
"case_id": case["id"],
"category": case["category"],
"correct_tool_called": correct_tool_called,
"tools_called": [tc["name"] for tc in tool_calls_made],
"passed": correct_tool_called,
}
finally:
# 元の実装を復元
for name, impl in original_implementations.items():
TOOL_IMPLEMENTATIONS[name] = impl
def run_evaluation(agent_fn) -> dict:
results = []
for case in eval_cases:
if case["category"] in ("factual_recall", "procedural"):
results.append(evaluate_factual_recall(agent_fn, case))
elif case["category"] == "tool_use":
results.append(evaluate_tool_use(agent_fn, case))
total = len(results)
passed = sum(1 for r in results if r["passed"])
return {
"total_cases": total,
"passed": passed,
"pass_rate": passed / total if total > 0 else 0,
"results": results,
}ステップ 3：LLM-as-Judge
# llm_judge.py
def llm_judge_evaluation(case: dict, agent_response: str) -> dict:
"""GPT-4 を審査員として回答品質を評価"""
judge_prompt = f"""You are an expert evaluator for an onboarding assistant.
User Question: {case['input']}
Expected Answer: {case['expected_output']}
Agent Response: {agent_response}
Rate the response on these dimensions (1-5 scale):
1. Accuracy: Is the information correct?
2. Completeness: Does it cover all key points?
3. Helpfulness: Would a new employee find this useful?
4. Tone: Is it professional and friendly?
Output a JSON object with scores and a brief explanation."""
response = client.chat.completions.create(
model="gpt-4",
messages=[
{"role": "system", "content": judge_prompt},
{"role": "user", "content": "Evaluate the response."},
],
)
return json.loads(response.choices[0].message.content)ステップ 4：結果に基づく反復改善
# iteration.py
def run_eval_improve_loop(max_iterations: int = 3):
for iteration in range(max_iterations):
print(f"\n=== Iteration {iteration + 1} ===")
# 評価を実行
results = run_evaluation(memory_aware_agent_chat)
print(f"Pass rate: {results['pass_rate']:.0%}")
# 失敗ケースを分析
failures = [r for r in results["results"] if not r["passed"]]
for f in failures:
print(f" FAIL [{f['case_id']}]: {f.get('response', '')[:100]}")
if results["pass_rate"] >= 0.9:
print("Target reached!")
break
# 反復：プロンプトの改善、ドキュメントの追加、ツールの修正
print(" → Improving system prompt and adding more documents...")
# （実際には、プロンプトの修正やドキュメントの追加などを行う）ステージ 5 完了：測定可能で反復的に改善される Agent システムが完成しました。最後のステップ：本番対応にすることです。
6.6 蒸留とデプロイ
問題
Agent は GPT-4 で素晴らしい動作をしますが、スケールするとコストが持続不可能です：
- 社員 500 名 × 1 日 10 クエリ × 30 日 = 月間 150,000 クエリ
- 1 クエリあたり約 2000 トークン（入力 + 出力）、月間 3 億トークン
- GPT-4 コスト：約 $3,000/月
特定のドメインで同等の性能を発揮する、より安価なモデルが必要です。
ステップ 1：小規模モデルへの蒸留
# distillation.py
# GPT-4 を教師モデルとしてトレーニングデータを生成
def generate_training_data(n_examples: int = 1000) -> list[dict]:
training_data = []
for case in eval_cases * (n_examples // len(eval_cases)):
# バリエーションを追加してより多様なデータを作成
variations = [
case["input"],
f"Hey, {case['input'].lower()}",
f"Quick question: {case['input']}",
]
for variant in variations:
response = client.chat.completions.create(
model="gpt-4",
messages=[
{"role": "system", "content": "You are an onboarding assistant for Acme Corp..."},
{"role": "user", "content": variant},
],
)
training_data.append({
"input": variant,
"output": response.choices[0].message.content,
})
return training_data
# トレーニングデータを使って小規模モデル（例：Llama 2 7B）をファインチューニング
# （詳細なファインチューニングコードは Production チャプターを参照）ステップ 2：モニタリングのセットアップ
# monitoring.py
from prometheus_client import Counter, Histogram, start_http_server
request_counter = Counter('qa_agent_requests_total', 'Total requests', ['intent', 'status'])
request_latency = Histogram('qa_agent_latency_seconds', 'Request latency')
token_counter = Counter('qa_agent_tokens_total', 'Token usage', ['type'])
# Prometheus メトリクスサーバーを起動
start_http_server(8000)
def tracked_agent_chat(user_id: str, user_message: str) -> str:
import time
start = time.time()
try:
response = memory_aware_agent_chat(user_id, user_message)
request_counter.labels(intent="general", status="success").inc()
return response
except Exception as e:
request_counter.labels(intent="general", status="error").inc()
raise
finally:
request_latency.observe(time.time() - start)ステップ 3：カナリアリリースでデプロイ
# deployment.py
def route_request(user_id: str, user_message: str) -> str:
"""トラフィックの 5% を新しい蒸留モデルにルーティング"""
import hashlib
bucket = int(hashlib.md5(user_id.encode()).hexdigest(), 16) % 100
if bucket < 5:
# カナリア：蒸留モデルを使用
return distilled_model_chat(user_id, user_message)
else:
# 安定版：GPT-4 を使用
return memory_aware_agent_chat(user_id, user_message)最終アーキテクチャ
┌─────────────────────────────────────────────────────────┐
│ Q&A Agent System │
│ │
│ User ──▶ Router ──▶ Agent Core ──▶ LLM (GPT-4/7B) │
│ │ │ │
│ │ ├── RAG Pipeline (Vector DB) │
│ │ ├── Tool Registry (HR API) │
│ │ ├── Memory Store (User Profile) │
│ │ └── Skill Loader (Onboarding) │
│ │ │
│ └── Monitoring (Prometheus + Grafana) │
│ │
│ Canary: 5% → Distilled Model (7B, fine-tuned) │
│ Stable: 95% → GPT-4 │
└─────────────────────────────────────────────────────────┘構築したもの
1 つの API コールから始まり、段階的に以下を構築しました：
| ステージ | 機能 | 適用した知識 |
|---|---|---|
| 6.1 | 基本的な会話 | LLM 基礎（トークン、コンテキストウィンドウ） |
| 6.2 | ドキュメントに基づく回答 | RAG（埋め込み、チャンキング、検索） |
| 6.3 | ツール使用とプランニング | Agent コア（Function Calling、ReAct、MCP） |
| 6.4 | メモリとパーソナライズされた回答 | メモリ＆スキルシステム |
| 6.5 | 測定可能な品質 | 評価フレームワーク |
| 6.6 | コスト効率の良い本番システム | 蒸留、モニタリング、カナリアリリース |
これが完全な Agent エンジニアリングのパスです——「Hello World」から本番環境まで。