npx skills add ...
npx skills add langchain-ai/skills-benchmarks --skill langchain-middleware
npx skills add langchain-ai/skills-benchmarks --skill langchain-middleware
INVOKE THIS SKILL when you need human-in-the-loop approval, custom middleware, or structured output. Covers HumanInTheLoopMiddleware for human approval of dangerous tool calls, creating custom middleware with hooks, Command resume patterns, and structured output with Pydantic/Zod.
Requirements: Checkpointer + thread_id config for all HITL workflows.
@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"]}, } ) ], )
config = {"configurable": {"thread_id": "session-1"}}
result1 = agent.invoke({ "messages": [{"role": "user", "content": "Send email to john@example.com"}] }, config=config)
if "interrupt" in result1: print(f"Waiting for approval: {result1['interrupt']}")
result2 = agent.invoke( Command(resume={"decisions": [{"type": "approve"}]}), config=config )
before_model, after_model, wrap_tool_call, before_agent, after_agentSix decorator hooks are available. Two patterns:
wrap_tool_call, wrap_model_call): (request, handler) — call handler(request) to proceed, or return early to short-circuit.before_model, after_model, before_agent, after_agent): (state, runtime) — inspect or modify state. Return None or a dict of state updates.agent = create_agent( model="gpt-4.1", tools=[send_email], checkpointer=MemorySaver(), # Required middleware=[HumanInTheLoopMiddleware({...})] )
agent.invoke(input, config={"configurable": {"thread_id": "user-123"}})
// CORRECT import { Command } from "@langchain/langgraph"; await agent.invoke(new Command({ resume: { decisions: [{ type: "approve" }] } }), config);
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");
},
});