langchain-middleware について
本番向け LangChain エージェントに人間の承認やカスタムミドルウェアを加えるスキルです。ツール呼び出しの一時停止、編集、承認、拒否に加え、ログやリトライのフックを扱います。
主な機能
- 01ツール呼び出しの人間承認
- 02ツールごとの判断ポリシー
- 03承認・拒否後の実行再開
- 04エージェントとツールのフック
- 05ログ、リトライ、エラー処理
活用シーン
- 01メール送信前に承認を求める
- 02実行前にツール引数を編集する
- 03高リスク操作を理由付きで拒否する
- 04ツール呼び出しにログやリトライを加える
インストール
コマンドをコピーしてターミナルで実行します。
$
npx skills add https:/インストール後、Claude Code で/langchain-middlewareを入力して読み込まれたか確認します。
詳細
本番環境の LangChain エージェント向けミドルウェアパターン：
- HumanInTheLoopMiddleware / humanInTheLoopMiddleware：危険なツール呼び出しの前に一時停止し、人間の承認を求めます
- カスタムミドルウェア：ツール呼び出しをインターセプトし、エラー処理、ロギング、再試行ロジックを実行します
- Command による再開：人間の判断（承認、編集、拒否）の後に実行を続行します
要件： すべての HITL ワークフローで、checkpointer と thread_id の設定が必要です。
Human-in-the-Loop
承認を得てからメールを送信するように一時停止する HITL ミドルウェアを使って、エージェントをセットアップします。 ```python from langchain.agents import create_agent from langchain.agents.middleware import HumanInTheLoopMiddleware from langgraph.checkpoint.memory import MemorySaver from langchain.tools import tool@tool def send_email(to: str, subject: str, body: str) -> str: """Send an email.""" return f"Email sent to {to}"
agent = create_agent( model="gpt-4.1", tools=[send_email], checkpointer=MemorySaver(), # Required for HITL middleware=[ HumanInTheLoopMiddleware( interrupt_on={ "send_email": {"allowed_decisions": ["approve", "edit", "reject"]}, } ) ], )
</python>
<typescript>
人間の承認を得てからメールを送信するように一時停止する HITL を使って、エージェントをセットアップします。
```typescript
import { createAgent, humanInTheLoopMiddleware } from "langchain";
import { MemorySaver } from "@langchain/langgraph";
import { tool } from "@langchain/core/tools";
import { z } from "zod";
const sendEmail = tool(
async ({ to, subject, body }) => `Email sent to ${to}`,
{
name: "send_email",
description: "Send an email",
schema: z.object({ to: z.string(), subject: z.string(), body: z.string() }),
}
);
const agent = createAgent({
model: "anthropic:claude-sonnet-4-5",
tools: [sendEmail],
checkpointer: new MemorySaver(),
middleware: [
humanInTheLoopMiddleware({
interruptOn: { send_email: { allowedDecisions: ["approve", "edit", "reject"] } },
}),
],
});
config = {"configurable": {"thread_id": "session-1"}}
Step 1: Agent runs until it needs to call tool
result1 = agent.invoke({ "messages": [{"role": "user", "content": "Send email to [email protected]"}] }, config=config)
Check for interrupt
if "interrupt" in result1: print(f"Waiting for approval: {result1['interrupt']}")
Step 2: Human approves
result2 = agent.invoke( Command(resume={"decisions": [{"type": "approve"}]}), config=config )
</python>
<typescript>
エージェントを実行し、中断を検出してから、人間の承認後に実行を再開します。
```typescript
import { Command } from "@langchain/langgraph";
const config = { configurable: { thread_id: "session-1" } };
// Step 1: Agent runs until it needs to call tool
const result1 = await agent.invoke({
messages: [{ role: "user", content: "Send email to [email protected]" }]
}, config);
// Check for interrupt
if (result1.__interrupt__) {
console.log(`Waiting for approval: ${result1.__interrupt__}`);
}
// Step 2: Human approves
const result2 = await agent.invoke(
new Command({ resume: { decisions: [{ type: "approve" }] } }),
config
);
- 承認が必要なツール（ツールごとのポリシー）
- ツールごとに許可する判断（承認、編集、拒否）
- カスタムミドルウェアフック：
before_model、after_model、wrap_tool_call、before_agent、after_agent - ツール固有のミドルウェア（特定のツールのみに適用）
カスタムミドルウェアフック
6 種類のデコレーターフックを利用できます。パターンは 2 種類です。
- ラップフック（
wrap_tool_call、wrap_model_call）：(request, handler)— 続行するにはhandler(request)を呼び出し、早期に返すと処理を短絡します。 - 前後フック（
before_model、after_model、before_agent、after_agent）：(state, runtime)— state を確認または変更します。Noneまたは state の更新を含む dict を返します。
from langchain.agents.middleware import wrap_tool_call
@wrap_tool_call
def retry_middleware(request, handler):
for attempt in range(3):
try:
return handler(request)
except Exception:
if attempt == 2:
raise
@wrap_tool_call
def guard_middleware(request, handler):
if request.tool_call["name"] == "dangerous_tool":
return "This tool is disabled" # short-circuit
return handler(request)
import { createMiddleware } from "langchain";
const retryMiddleware = createMiddleware({
wrapToolCall: async (request, handler) => {
for (let attempt = 0; attempt < 3; attempt++) {
try { return await handler(request); }
catch (e) { if (attempt === 2) throw e; }
}
},
});
from langchain.agents.middleware import before_model, after_model
@before_model
def log_calls(state, runtime):
print(f"Calling model with {len(state['messages'])} messages")
@after_model
def check_output(state, runtime):
print(f"Model responded")
import { createMiddleware } from "langchain";
const loggingMiddleware = createMiddleware({
beforeModel: (state, runtime) => {
console.log(`Calling model with ${state.messages.length} messages`);
},
afterModel: (state, runtime) => {
console.log("Model responded");
},
});
- ツール実行後の中断（実行前に行う必要があります）
- HITL に必要な checkpointer の省略
CORRECT
agent = create_agent( model="gpt-4.1", tools=[send_email], checkpointer=MemorySaver(), # Required middleware=[HumanInTheLoopMiddleware({...})] )
</python>
<typescript>
HITL で state を永続化するには、checkpointer が必要です。
```typescript
// WRONG: No checkpointer
const agent = createAgent({
model: "anthropic:claude-sonnet-4-5", tools: [sendEmail],
middleware: [humanInTheLoopMiddleware({ interruptOn: { send_email: true } })],
});
// CORRECT: Add checkpointer
const agent = createAgent({
model: "anthropic:claude-sonnet-4-5", tools: [sendEmail],
checkpointer: new MemorySaver(),
middleware: [humanInTheLoopMiddleware({ interruptOn: { send_email: true } })],
});
CORRECT
agent.invoke(input, config={"configurable": {"thread_id": "user-123"}})
</python>
</fix-no-thread-id>
<fix-wrong-resume-syntax>
<python>
中断後に実行を再開するには、Command クラスを使用します。
```python
# WRONG
agent.invoke({"resume": {"decisions": [...]}})
# CORRECT
from langgraph.types import Command
agent.invoke(Command(resume={"decisions": [{"type": "approve"}]}), config=config)
// CORRECT import { Command } from "@langchain/langgraph"; await agent.invoke(new Command({ resume: { decisions: [{ type: "approve" }] } }), config);
</typescript>
</fix-wrong-resume-syntax>
コメント
Markdown 対応