llp

你将学会： 在不改循环的前提下限制模型乱转；理解 HITL 必须有 checkpointer。
前置： L02、L03。

中间件 = 插在循环钩子上的插件：限流、重试、脱敏、改提示词。

flowchart LR
  BA[before_agent] --> BM[before_model]
  BM --> M[Model]
  M --> AM[after_model]
  AM --> T[Tool]
  T --> BM

1. 限制模型调用次数（需要 API key）

run_limit=1：模型只能想一次。它若选择调工具，工具会执行，但不会再有第二次「总结」。

from dotenv import load_dotenv

load_dotenv()

from langchain.agents import create_agent
from langchain.agents.middleware import ModelCallLimitMiddleware
from langchain.chat_models import init_chat_model

model = init_chat_model(
    "deepseek:deepseek-chat",
    temperature=0.5,
    max_tokens=2500,
    timeout=300,
    max_retries=6,
)


def get_weather(city: str) -> str:
    """获取某城市的天气。"""
    return f"{city}今天晴，25°C。"


agent = create_agent(
    model=model,
    tools=[get_weather],
    middleware=[ModelCallLimitMiddleware(run_limit=1, exit_behavior="end")],
)
result = agent.invoke({
    "messages": [{"role": "user", "content": "请使用 get_weather 查询北京天气。"}]
})
for m in result["messages"]:
    print(m.type, bool(getattr(m, "tool_calls", None)), m.text[:50])

2. 人在回路（不需要 API key）

敏感动作先停住。没有 checkpointer 会报 MISSING_CHECKPOINTER。

下面用了一张最小的图。看不懂节点和边没关系：先记住「interrupt 必须有 checkpointer」。图画法在 L10。

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, interrupt
from typing_extensions import TypedDict


class State(TypedDict):
    draft: str
    approved: bool


def review(state: State):
    decision = interrupt({"draft": state["draft"], "ask": "approve?"})
    ok = bool(decision)
    return {"approved": ok, "draft": state["draft"] if ok else "rejected"}


app = (
    StateGraph(State)
    .add_node("review", review)
    .add_edge(START, "review")
    .add_edge("review", END)
    .compile(checkpointer=InMemorySaver())
)
cfg = {"configurable": {"thread_id": "hitl-1"}}
paused = app.invoke({"draft": "删除全部订单", "approved": False}, config=cfg)
print("paused =>", paused.get("__interrupt__"))
print("resumed =>", app.invoke(Command(resume=True), config=cfg))

PII 脱敏、重试、Context editing 见仓库 08-中间件/、03-代理与工具/Guardrails护栏.py。

练习

1. 把 run_limit 改成 5，再跑天气问题，对比轨迹长度。
2. HITL 脚本第二次传入 Command(resume=False)，确认 draft 变成 rejected。

本章验收

•  能给 create_agent 挂一个内置中间件
•  能解释 HITL 为什么必须 checkpointer

下一课：流式输出