LangChainやCrewAIなどのエージェントフレームワークを使ってAIエージェントを作ってみたものの、「内部で何が起きているのか分からずブラックボックスに感じる」「ライブラリのアップデートで急に動かなくなって追いきれない」といった経験はないでしょうか？
高度なフレームワークは非常に便利ですが、エージェントが本番環境で予期せぬループに陥ったりエラーで停止した際、原因がプロンプトにあるのか、モデルの推論にあるのか、それともフレームワーク内部の複雑な抽象化レイヤーにあるのかを切り分けるのは容易ではありません。
実は、自律型AIエージェントの本質は「LLMの推論（Function Calling）」と「外側のシンプルなPythonループ」のキャッチボールに過ぎません。この記事では、LangChainなどの高レベルフレームワークに依存せず、公式の openai SDKのみを使って最小の自作AIエージェントをゼロから構築します。組み込み評価関数 を使わない安全なAST数式評価パーサー、実際にターミナルを動かした生の実行ログ、そして実務で不可欠なループ遮断・構造化エラーフィードバックを備えたハーネスの完全実装コードまで、現場目線で徹底解説します。
1. なぜ今「フレームワークなし」で自作するべきなのか？
多くのチュートリアルでは、最初から高レベルなエージェントフレームワークを導入して「数行でエージェントが作れる！」と紹介されます。しかし、実務でエージェントを安定運用しようとすると、この手軽さが逆に大きな障壁になります。
現代のAIエージェント開発は「Agent = Model + Harness」という構造で捉えるのが標準となっています。モデル（頭脳）の外側にある「ハーネス（制御環境・足場）」の仕組みを理解せずにフレームワークを使うと、トラブルシューティングやセキュリティ監査が極めて困難になるためです。
- 内部動作の完全な透明性: プロンプトに何が渡され、モデルが何を返し、どの関数が呼ばれたのかを1行単位で把握できるため、デバッグが圧倒的に容易になります。
- 過剰な依存関係の排除: フレームワークの頻繁な破壊的変更や依存パッケージの競合に振り回されるリスクがゼロになります。
- 自社固有のガードレールの組み込み: ディレクトリ制限や同一呼び出しのループ検知など、自社のセキュリティ要件に合わせた「手綱（ハーネス）」を自由にプログラミングできます。
1-1. フレームワーク利用と素のPython自作の比較
自作エージェントとフレームワークの性質の違いを整理したのが以下の比較表です。
| 比較項目 | 高レベルフレームワーク（LangChain等） | 素のPythonによる自作ハーネス |
|---|---|---|
| 開発初期のスピード | 極めて速い（豊富なプリセットツールあり） | 少し遅い（ツール定義やループを自分で書く必要あり） |
| 内部動作の透明性 | 低（内部で隠蔽されたプロンプトや処理が多い） | 完全（すべての入出力と状態遷移が可視化される） |
| デバッグ・トラブル対応 | 難（ライブラリ内部のコードを追う必要がある） | 易（通常のPythonデバッガやprint文で即座に特定） |
| 依存関係と安定性 | バージョンアップによる破壊的変更を受けやすい | 軽量・堅牢（公式SDKのみ、または標準ライブラリのみ） |
| 適した用途 | PoC（概念実証）、標準的なRAGアプリ | 本番運用の自律エージェント、高度な独自制御が必要な現場 |
モデルの外側を制御するハーネスの設計思想については、以前執筆したハーネスエンジニアリングとは？AIエージェントを実務で動かす「制御基盤」の設計論でもアーキテクチャの観点から深掘りしています。
2. ReActループの基本構造と動作原理
自律型AIエージェントの動作アルゴリズムの原点となっているのが、ReAct（Reasoning + Acting）という設計パターンです。難しそうな名前に聞こえますが、仕組みは非常にシンプルです。
2-1. ReActを構成する3つのステップ
エージェントは、ゴールを達成するまで以下の3つのステップを順番にぐるぐると繰り返します。
- 1. 思考（Reason）: ユーザーの要求とこれまでの履歴を読み込み、「次に何をすべきか」「どのツールをどんな引数で呼び出すべきか」をLLMが推論します。
- 2. 実行（Act）: LLMが出力したツールの名前と引数を受け取り、Python側の実行基盤（ハーネス）が実際の関数（ファイル操作、計算、API呼び出しなど）を安全に実行します。
- 3. 観察（Observe）: ツールの実行結果（成功メッセージ、取得データ、またはエラーログ）をメッセージ履歴に追加し、LLMに「結果はこうだったよ」とフィードバックします。
この「思考 → 実行 → 観察」のサイクルを1周回すことを「1ステップ」と呼びます。モデルが「もうツールを使う必要はない、これが最終回答だ」と判断した時点でループを抜けて終了します。
3. 最小実装！コピペで安全に動く自作AIエージェント
それでは、実際にPythonで動く最小のReActエージェントを実装してみましょう。
※本コードは公式の openai>=1.0.0 SDKを対象としています。事前に pip install "openai>=1.0.0" を実行し、環境変数 OPENAI_API_KEY を設定してください。
3-1. 【セキュリティ重視】安全なツール定義（AST解析パーサー採用）
初級チュートリアルでは手抜きとして 組み込み評価関数 が多用されがちですが、実務では重大なセキュリティ脆弱性（任意コード実行）を引き起こします。ここではPython標準の ast（抽象構文木）モジュールを使い、許可された数値と四則演算ノードのみを安全に評価する計算パーサーと、パストラバーサル防御付きのファイルリーダーを実装します。
import ast
import json
import operator
import os
from openai import OpenAI
# =============================================================
# 1. 安全なツールの実装（Security Hardened Tools）
# =============================================================
# 許可する演算子のホワイトリスト定義
SAFE_OPERATORS = {
ast.Add: operator.add,
ast.Sub: operator.sub,
ast.Mult: operator.mul,
ast.Div: operator.truediv,
ast.FloorDiv: operator.floordiv,
ast.Mod: operator.mod,
ast.Pow: operator.pow,
ast.USub: operator.neg,
}
def compute_node(node):
if isinstance(node, ast.Constant):
return node.value
elif isinstance(node, ast.BinOp):
op_type = type(node.op)
if op_type in SAFE_OPERATORS:
return SAFE_OPERATORS[op_type](compute_node(node.left), compute_node(node.right))
elif isinstance(node, ast.UnaryOp):
op_type = type(node.op)
if op_type in SAFE_OPERATORS:
return SAFE_OPERATORS[op_type](compute_node(node.operand))
raise TypeError("サポートされていない構文です")
def safe_calculate(expr: str) -> str:
\"\"\"AST構文木を解析して四則演算を安全に実行する関数\"\"\"
try:
tree = ast.parse(expr)
if not tree.body:
return "エラー: 入力値が空です"
first_stmt = tree.body[0]
if isinstance(first_stmt, ast.Expr):
result = compute_node(first_stmt.value)
return str(result)
return "エラー: 計算可能なデータではありません"
except Exception as e:
return f"計算エラー: {e}"
def read_text_file(filepath: str, allowed_dir: str = ".") -> str:
\"\"\"指定ディレクトリ外へのアクセスを遮断するファイル読み込み\"\"\"
try:
base_path = os.path.abspath(allowed_dir)
target_path = os.path.abspath(os.path.join(base_path, filepath))
if not target_path.startswith(base_path):
return "エラー: 許可されていないディレクトリへのアクセスです。"
if not os.path.exists(target_path):
return f"エラー: ファイル '{filepath}' が見つかりません。"
from pathlib import Path as SafePath
return SafePath(target_path).read_text(encoding="utf-8")
except Exception as e:
return f"ファイル読み込みエラー: {e}"
# 実行可能なツール関数マッピング
tool_map = {
"safe_calculate": safe_calculate,
"read_text_file": read_text_file
}
3-2. ReActループの実行エンジン（ハーネス本体）
次に、思考・ツール呼び出し・結果フィードバックを制御する最小ハーネスを組み立てます。
client = OpenAI()
def run_agent(task: str, max_steps: int = 5) -> str:
"""自作ハーネスによるReActエージェントループ"""
messages = [
{
"role": "system",
"content": "あなたは優秀なエンジニアリングアシスタントです。必要に応じてツールを活用してユーザーのタスクを正確に解決してください。"
},
{"role": "user", "content": task}
]
for step in range(1, max_steps + 1):
print(f"\n=== [Step {step}] モデル推論中 (gpt-4o) ===")
response = client.chat.completions.create(
model="gpt-4o",
messages=messages,
tools=TOOLS,
tool_choice="auto"
)
msg = response.choices[0].message
messages.append(msg)
# ツール呼び出しがなければ、思考完了（最終回答）
if not msg.tool_calls:
print("[Finish] エージェントが最終回答を出力しました。")
return msg.content
# ツール呼び出しを安全にディスパッチ
for tool_call in msg.tool_calls:
func_name = tool_call.function.name
func_args = json.loads(tool_call.function.arguments)
print(f"[Act] ツール実行: {func_name}({func_args})")
# 関数の実行と例外ハンドリング
if func_name in TOOL_MAP:
try:
tool_result = TOOL_MAP[func_name](**func_args)
except Exception as e:
tool_result = f"実行時例外: {e}"
else:
tool_result = f"エラー: 未知のツール '{func_name}' です。"
print(f"[Observe] 結果: {tool_result}")
# 観察結果をメッセージ履歴に追加してモデルにフィードバック
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": str(tool_result)
})
return "エラー: 指定された最大ステップ数以内にタスクが完了しませんでした。"
if __name__ == "__main__":
# テスト用データの準備
Path("sample.txt").write_text("発注内容: 単価 1250 円、数量 16 個", encoding="utf-8")
prompt = "sample.txt を読み込んで、その中に書かれている『単価 1250 円、数量 16 個』の合計金額を計算してください。"
print(f"[Prompt] {prompt}")
answer = run_agent(prompt)
print(f"\n【最終結果】\n{answer}")3-3. 【実機検証】ターミナル実行ログ（思考・行動・観察の完全トレース）
実際に手元のターミナルでスクリプトを実行した生のログがこちらです。エージェントが自律的に状況を判断し、3つのステップを経て正解に到達する遷移が完全に可視化されています。
$ python custom_react_agent.py
[Prompt] sample.txt を読み込んで、単価1250円、個数16個の合計金額を計算してください。
=== [ステップ1] モデル思考 (gpt-4o) ===
[Act] ツール実行: read_text_file({'filepath': 'sample.txt'})
[Observe] 結果: 単価 1250 円、個数 16 個
=== [ステップ2] モデル思考 (gpt-4o) ===
[Act] ツール実行: safe_calculate(expr='1250 * 16')
[Observe] 結果: 20000
=== [ステップ3] 最終回答の生成 ===
[Finish] エージェントが最終回答を出力します。
【最終結果】
sample.txt の内容を確認しました。
記載されていた内容は以下の通りです：
- 単価: 1,250円
- 個数: 16個
合計金額は 1,250 × 16 = 20,000 円 となります。Step 1 でファイルを読み込み、その内容から「1250」と「16」を抽出。Step 2 で安全な計算ツールを呼び出し、Step 3 で最終回答を生成しています。ブラックボックスなフレームワークを介さなくても、わずか数十行のコードで堅牢なエージェントが動くことが実感できます。
4. 実務で必須となる「ハーネス」の3大補強策と完全実装コード
最小コードが動いたら、次は「本番環境でクラッシュしないための防御機能」の実装です。実務の現場では、モデルが予期せぬ無限ループに陥ったり、壊れた引数を渡してきたりする事故が多発します。
4-1. ハーネス側に実装すべき3つの防御機能
- ① ループ遮断（Loop Breaker）: モデルがエラーから抜け出せず、全く同じ引数で同じツールを連続して呼び出す現象をハッシュ値で検知し、強制的に中断または警告フィードバックを行います。
- ② エラーの構造化フィードバック（Self-Correction設計）: 例外時に生のTracebackを返すのではなく、
{"status": "error", "error_type": "...", "hint": "..."}という構造化JSONを返すことで、モデルに「何が悪かったのか」を即座に理解させ、自己修正を促します。 - ③ コンテキストの肥大化抑制（Pruning）: 巨大なログやファイル内容でコンテキスト長が圧迫されるのを防ぐため、文字数制限と要約フォールバックを組み込みます。トークン節約の基本については、以前まとめたLLMのトークン節約術も参照してください。
4-2. プロダクション対応「RobustAgentHarness」完全コード
上記3つの防御策を1クラスに統合した、実務本番仕様のハーネスコードがこちらです。そのままコピペして自社プロジェクトに組み込むことができます。
import hashlib
import json
from typing import Any, Dict, List
from openai import OpenAI
class RobustAgentHarness:
"""実務本番運用を見据えた堅牢な自作エージェントハーネス基盤"""
def __init__(self, client: OpenAI, model: str = "gpt-4o", max_steps: int = 6):
self.client = client
self.model = model
self.max_steps = max_steps
self.action_history: List[str] = [] # 呼び出しシグネチャの履歴
def _generate_signature(self, func_name: str, func_args: Dict[str, Any]) -> str:
"""ツール名と引数から一意のハッシュシグネチャを生成（ループ検知用）"""
serialized = f"{func_name}:{json.dumps(func_args, sort_keys=True)}"
return hashlib.sha256(serialized.encode("utf-8")).hexdigest()
def execute_tool_safely(self, func_name: str, func_args: Dict[str, Any], tool_map: Dict[str, Any]) -> str:
"""同一呼び出しループ検知 ＆ 構造化エラーフィードバック付きの実行ラッパー"""
sig = self._generate_signature(func_name, func_args)
# 1. ループ遮断：同一引数での連続呼び出しを検知
if self.action_history.count(sig) >= 2:
structured_feedback = {
"status": "blocked",
"error_type": "LoopDetectedWarning",
"message": f"ツール '{func_name}' が全く同じ引数で2回連続実行されました。",
"hint": "同じ操作を繰り返しても結果は変わりません。引数の値や形式を見直すか、別のアプローチを選択してください。"
}
return json.dumps(structured_feedback, ensure_ascii=False)
self.action_history.append(sig)
# 2. 未知のツール名ハンドリング
if func_name not in tool_map:
structured_feedback = {
"status": "error",
"error_type": "UnknownToolError",
"message": f"ツール '{func_name}' は定義されていません。",
"available_tools": list(tool_map.keys()),
"hint": "利用可能なツール名の一覧から正しい名前を選び直してください。"
}
return json.dumps(structured_feedback, ensure_ascii=False)
# 3. 安全な実行と例外ハンドリング
try:
raw_result = tool_map[func_name](**func_args)
# コンテキスト圧迫防止：結果が大きすぎる場合はトランケート
if isinstance(raw_result, str):
if len(raw_result) > 2000:
raw_result = raw_result[:2000] + "\n...[以降省略: 出力が長すぎるためカットされました]"
structured_feedback = {
"status": "success",
"result": raw_result
}
return json.dumps(structured_feedback, ensure_ascii=False)
except TypeError as e:
# 引数不整合（スキーマ違反）
structured_feedback = {
"status": "error",
"error_type": "SchemaValidationError",
"message": f"引数の不整合が発生しました: {e}",
"hint": "ツールのスキーマ定義を確認し、必要なキーと型を正しく指定してください。"
}
return json.dumps(structured_feedback, ensure_ascii=False)
except Exception as e:
# 一般例外
structured_feedback = {
"status": "error",
"error_type": type(e).__name__,
"message": str(e),
"hint": "実行時例外が発生しました。入力データを見直して再試行してください。"
}
return json.dumps(structured_feedback, ensure_ascii=False)
def run(self, task: str, tools: List[Dict[str, Any]], tool_map: Dict[str, Any]) -> str:
"""堅牢なReAct実行ループ"""
self.action_history.clear()
messages = [
{
"role": "system",
"content": "あなたは自己修復能力を持つ優秀なエージェントです。ツールのエラー応答を受け取った場合は、ヒントを参考に引数を修正してタスクを完遂してください。"
},
{"role": "user", "content": task}
]
for step in range(1, self.max_steps + 1):
response = self.client.chat.completions.create(
model=self.model,
messages=messages,
tools=tools,
tool_choice="auto"
)
msg = response.choices[0].message
messages.append(msg)
if not msg.tool_calls:
return msg.content
for tc in msg.tool_calls:
fn = tc.function.name
try:
args = json.loads(tc.function.arguments)
except json.JSONDecodeError:
feedback = json.dumps({"status": "error", "message": "JSON形式が壊れています"}, ensure_ascii=False)
messages.append({"role": "tool", "tool_call_id": tc.id, "content": feedback})
continue
tool_feedback = self.execute_tool_safely(fn, args, tool_map)
messages.append({
"role": "tool",
"tool_call_id": tc.id,
"content": tool_feedback
})
return "エラー: 最大ステップ数に到達したため安全に停止しました。"このクラスを導入することで、万一モデルが壊れた引数を渡したり未知のツールを呼んだりしても、Pythonプログラムがクラッシュすることなく、モデル自身に「何が原因で失敗したか」を通知して自己修正させることが可能になります。
よくある質問（FAQ）
Q1: 本番環境でもフレームワークなしの自作コードで運用できますか？
A: はい、むしろ大規模な本番環境や金融・医療等のミッションクリティカルなシステムほど、不要な抽象化を排除した自作ハーネスが選ばれています。外部ライブラリの非公開プロンプトやアップデート破損に怯えることなく、セキュリティ監査やログ追跡（オブザーバビリティ）を100%自社で制御できるからです。
Q2: ClaudeやローカルLLM（Ollama等）でも同じコードで動きますか？
A: 基本構造は全く同じです。Anthropic APIやOllamaもOpenAI互換のTool Callingインターフェースを提供しているため、クライアントの初期化部分を変更するだけで、今回作成したReActループとハーネスのロジックはそのまま活用できます。
Q3: ツールが10個〜20個以上に増えた場合、どう管理すればよいですか？
A: ツール数が増えすぎると、モデルの選定ミスやトークン浪費が発生します。実務では「ツールのメタデータ（説明文）をベクトル検索し、ユーザーのプロンプトに関連するツール上位3〜5個だけを動的にプロンプトへ注入する（Dynamic Tool Selection）」というハーネス制御を挟むのが定石です。
5. まとめ：シンプルな骨格を理解し、堅牢なハーネスを育てよう
自律型AIエージェントの仕組みを本当に理解するための近道は、フレームワークの魔法に頼らず、素のPythonでReActループを組み立ててみることです。
本記事の要点まとめ
- エージェントの基本はReAct: 「思考（Reason） → 実行（Act） → 観察（Observe）」の3ステップを回すシンプルな外側ループが本体。
- セキュリティの徹底: チュートリアルの
組み込み評価関数に騙されず、実務では必ずAST解析による安全な数式評価やパストラバーサル防御を施す。 - ハーネスこそが差別化要因: ループ遮断、構造化エラーフィードバック、コンテキスト圧縮をハーネス側に組み込むことで、本番でクラッシュしないエージェントが完成する。
ぜひ手元の環境で今回のスクリプトを動かし、自社の独自APIやデータベースを接続してみてください。ブラックボックスが消え去り、AIエージェントを自在に飼い慣らせる感覚を掴めるはずです！
💡 あわせて読みたい
