LangChain源码学习路线：第 15 篇 Middleware / Guardrails 与 Agent 执行控制源码解剖#

1
create_agent / ReAct 主循环
2
  ↓
3
Middleware / Guardrails
4
  ↓
5
Agentic RAG / Retrieval Runtime
6
  ↓
7
工程化 Agent 能力

1
from langchain.agents import create_agent, AgentMiddleware
2
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage
3

4
class TravelGuardrail(AgentMiddleware):
5
    def before_model(self, state, runtime):
6
        last_user = state["messages"][-1]
7
        if asks_for_unsafe_booking(last_user.content):
8
            return {"messages": [AIMessage(content="我不能直接替你完成高风险预订。")]}
9
        return None
10

11
    def wrap_model_call(self, request, handler):
12
        if request.state.get("trip_complexity") == "high":
13
            request.model = strong_model
14
        response = handler(request)
15
        return validate_model_response(response)
16

17
    def wrap_tool_call(self, request, handler):
18
        if request.tool.name == "book_hotel" and not request.state.get("user_confirmed"):
19
            return ToolMessage(
20
                content="需要用户确认后才能调用 book_hotel。",
21
                tool_call_id=request.tool_call["id"],
22
            )
23
        return handler(request)
24

25
agent = create_agent(
26
    model=default_model,
27
    tools=[search_flights, book_hotel],
28
    middleware=[TravelGuardrail()],
29
)
30

31
result = agent.invoke({
32
    "messages": [HumanMessage(content="帮我直接订今晚最便宜的酒店")]
33
})

1
create_agent(model, tools, middleware=[TravelGuardrail()])
2
  ↓
3
归一化 middleware 实例列表
4
  ↓
5
检查每个实例实现了哪些 hook：before_model / after_model / wrap_model_call / wrap_tool_call
6
  ↓
7
构造 model 节点、tools 节点和 hook wrapper stack
8
  ↓
9
编译为 Agent graph
10
  ↓
11
agent.invoke(input)
12
  ↓
13
before_model(state, runtime)
14
  ↓
15
wrap_model_call(ModelRequest, handler)
16
  ↓
17
AIMessage 可能包含 tool_calls
18
  ↓
19
wrap_tool_call(ToolCallRequest, handler)
20
  ↓
21
after_model(state, runtime) 或下一轮循环

1
def run_model_with_middleware(state, runtime, middleware_stack):
2
    update = run_before_model_hooks(middleware_stack, state, runtime)
3
    if update is not None:
4
        return update
5

6
    request = ModelRequest(
7
        model=runtime.model,
8
        messages=state["messages"],
9
        state=state,
10
        runtime=runtime,
11
    )
12

13
    response = call_wrapped_model(request, middleware_stack, base_model_handler)
14
    checked_update = run_after_model_hooks(middleware_stack, response, state, runtime)
15
    return checked_update or {"messages": [response]}

1
langgraph/
2
├── graph/
3
├── pregel/
4
├── types.py
5
└── func/
6

7
langchain_core/
8
├── runnables/
9
├── tools/
10
├── retrievers.py
11
└── messages/

1
1. 先读官方概念文档，确认公共契约。
2
2. 再读公开 API 或装饰器入口。
3
3. 顺着构建期对象进入 runtime。
4
4. 追踪一次 invoke / stream / tool call / task call。
5
5. 单独检查异常、重试、恢复和扩展点。
6
6. 最后回到旅行规划助手做工程判断。

1
Public API / Decorator / Tool Protocol
2
  ↓
3
Runtime wrapper
4
  ↓
5
State / Config / Context
6
  ↓
7
Storage / Model / Tool / Graph Runtime

1
公共 API
2
  负责：给业务代码稳定入口。
3
  不负责：暴露所有内部调度细节。
4

5
运行时协议
6
  负责：让状态、配置、工具、模型和持久化协作。
7
  不负责：替业务判断所有策略。
8

9
扩展接口
10
  负责：允许业务插入可维护的定制逻辑。
11
  不负责：保证错误扩展仍然安全。

1
先找公开入口
2
  ↓
3
确认输入输出类型
4
  ↓
5
沿调用方追到核心实现
6
  ↓
7
记录状态与对象变化
8
  ↓
9
检查分支、异常和结束条件
10
  ↓
11
回到设计目的

1
def create_travel_agent():
2
    middleware = [
3
        TravelGuardrail(policy=travel_policy),
4
        DynamicModelMiddleware(default=cheap_model, fallback=strong_model),
5
        ToolErrorMiddleware(max_retries=2),
6
    ]
7
    return create_agent(
8
        model=cheap_model,
9
        tools=[search_flights, book_hotel],
10
        middleware=middleware,
11
    )

1
def normalize_middleware(middleware):
2
    normalized = []
3
    for item in middleware or []:
4
        if isinstance(item, type):
5
            item = item()
6
        if not isinstance(item, AgentMiddleware):
7
            raise TypeError("middleware must be AgentMiddleware")
8
        normalized.append(item)
9
    return normalized

1
def collect_agent_hooks(middleware_stack):
2
    hooks = {
3
        "before_model": [],
4
        "after_model": [],
5
        "wrap_model_call": [],
6
        "wrap_tool_call": [],
7
    }
8
    for middleware in middleware_stack:
9
        for hook_name in hooks:
10
            hook = getattr(middleware, hook_name, None)
11
            if hook is not None and is_overridden(middleware, hook_name):
12
                hooks[hook_name].append(hook)
13
    return hooks

1
class TravelPolicyGuardrail(AgentMiddleware):
2
    def __init__(self, blocked_tools: set[str], requires_confirmation: set[str]):
3
        self.blocked_tools = blocked_tools
4
        self.requires_confirmation = requires_confirmation
5

6
    def wrap_tool_call(self, request, handler):
7
        tool_name = request.tool.name
8
        if tool_name in self.blocked_tools:
9
            return make_denied_tool_message(request, reason="tool_blocked")
10
        if tool_name in self.requires_confirmation and not request.state.get("user_confirmed"):
11
            return make_denied_tool_message(request, reason="confirmation_required")
12
        return handler(request)

1
agent.invoke({"messages": [...]})
2
  ↓
3
恢复 / 初始化 agent state
4
  ↓
5
before_model(state, runtime)
6
  ↓
7
构造 ModelRequest(messages, model, state, runtime)
8
  ↓
9
wrap_model_call(request, handler)
10
  ↓
11
ChatModel 返回 AIMessage
12
  ↓
13
after_model(response/state, runtime)
14
  ↓
15
如果 AIMessage.tool_calls 非空，构造 ToolCallRequest
16
  ↓
17
wrap_tool_call(request, handler)
18
  ↓
19
ToolMessage 写回 state，进入下一轮或结束

1
def run_before_model_hooks(hooks, state, runtime):
2
    for hook in hooks["before_model"]:
3
        update = hook(state, runtime)
4
        if update is not None:
5
            validate_state_update(update)
6
            return update
7
    return None

1
def call_wrapped_model(request, hooks, base_handler):
2
    handler = base_handler
3
    for wrap in reversed(hooks["wrap_model_call"]):
4
        next_handler = handler
5
        handler = lambda req, wrap=wrap, next_handler=next_handler: wrap(req, next_handler)
6

7
    response = handler(request)
8

9
    for hook in hooks["after_model"]:
10
        update = hook(response, request.state, request.runtime)
11
        if update is not None:
12
            validate_state_update(update)
13
            return update
14

15
    return {"messages": [response]}

1
def run_tool_with_guardrails(tool_call, state, runtime, hooks):
2
    request = ToolCallRequest(
3
        tool=runtime.tools[tool_call["name"]],
4
        tool_call=tool_call,
5
        state=state,
6
        runtime=runtime,
7
    )
8

9
    def base_handler(req):
10
        result = req.tool.invoke(req.tool_call["args"])
11
        return ToolMessage(content=str(result), tool_call_id=req.tool_call["id"])
12

13
    handler = base_handler
14
    for wrap in reversed(hooks["wrap_tool_call"]):
15
        next_handler = handler
16
        handler = lambda req, wrap=wrap, next_handler=next_handler: wrap(req, next_handler)
17

18
    return handler(request)

1
def guarded_model_call(request, handler):
2
    try:
3
        return handler(request)
4
    except RateLimitError:
5
        request.model = request.runtime.fallback_model
6
        return handler(request)
7
    except PolicyViolation as error:
8
        return AIMessage(content=f"请求被策略拒绝：{error.reason}")
9
    except Exception:
10
        request.runtime.callbacks.on_error(...)
11
        raise

1
State
2
  ↓
3
before_model 可短路
4
  ↓
5
ModelRequest 可被 wrap_model_call 改写
6
  ↓
7
AIMessage 可被 after_model 校验
8
  ↓
9
ToolCallRequest 可被 wrap_tool_call 拦截
10
  ↓
11
ToolMessage / AIMessage 写回 state

1
def deny_unsafe_booking_before_model(state, runtime):
2
    user_text = state["messages"][-1].content
3
    if asks_to_book_without_confirmation(user_text):
4
        return {
5
            "messages": [AIMessage(content="我可以帮你比较选项，但需要你确认后才能预订。")],
6
            "policy_decisions": [{"policy": "booking_confirmation", "action": "blocked"}],
7
        }
8
    return None

1
def redact_and_route_model(request, handler):
2
    request.messages = redact_sensitive_blocks(request.messages)
3
    if requires_stronger_reasoning(request.state):
4
        request.model = request.runtime.models["strong"]
5

6
    response = handler(request)
7
    response.content = remove_internal_policy_text(response.content)
8
    return response

1
def guard_tool_call(request, handler):
2
    tool_name = request.tool.name
3
    args = request.tool_call["args"]
4

5
    if tool_name == "book_hotel" and not request.state.get("user_confirmed"):
6
        return ToolMessage(
7
            content="已拦截：book_hotel 需要用户确认。",
8
            tool_call_id=request.tool_call["id"],
9
        )
10

11
    try:
12
        return handler(request)
13
    except TimeoutError:
14
        return ToolMessage(
15
            content="工具超时，请稍后重试或改用搜索结果。",
16
            tool_call_id=request.tool_call["id"],
17
        )

1
def fallback_model_on_transient_error(request, handler):
2
    try:
3
        return handler(request)
4
    except RateLimitError:
5
        if request.runtime.fallback_model is None:
6
            raise
7
        downgraded = request.copy(update={"model": request.runtime.fallback_model})
8
        response = handler(downgraded)
9
        response.response_metadata["fallback_reason"] = "rate_limit"
10
        return response

1
正常结束：hook 放行，模型或工具返回协议对象，state update 被合并。
2
安全拒绝：before/after/tool guardrail 返回安全消息，并阻止高风险 handler 继续执行。
3
可恢复降级：限流、超时或临时失败进入 retry/fallback，并记录原因。
4
不可恢复失败：hook 返回非法对象、state 损坏、未知异常继续向上抛出并进入 trace。

1
class BudgetMiddleware(AgentMiddleware):
2
    def __init__(self, max_tool_calls: int):
3
        self.max_tool_calls = max_tool_calls
4

5
    def before_model(self, state, runtime):
6
        used = state.get("tool_call_count", 0)
7
        if used >= self.max_tool_calls:
8
            return {"messages": [AIMessage(content="本轮工具调用次数已达上限。")]}
9
        return None
10

11
    def wrap_tool_call(self, request, handler):
12
        result = handler(request)
13
        request.state["tool_call_count"] = request.state.get("tool_call_count", 0) + 1
14
        return result

1
def trace_policy_decision(runtime, request, decision):
2
    runtime.config["metadata"] = {
3
        **runtime.config.get("metadata", {}),
4
        "policy_name": decision.policy,
5
        "policy_action": decision.action,
6
        "tool_name": getattr(request.tool, "name", None),
7
    }
8

9

10
def guarded_tool_call(request, handler):
11
    decision = evaluate_tool_policy(request)
12
    trace_policy_decision(request.runtime, request, decision)
13

14
    if decision.action == "deny":
15
        return ToolMessage(content=decision.reason, tool_call_id=request.tool_call["id"])
16
    return handler(request)

1
class UserPolicyMiddleware(AgentMiddleware):
2
    def wrap_tool_call(self, request, handler):
3
        user_id = request.runtime.config["configurable"]["user_id"]
4
        namespace = ("travel_assistant", user_id, "tool_policy")
5
        policy_items = request.runtime.store.search(namespace, query=request.tool.name, limit=3)
6

7
        if denies_tool(policy_items, request.tool.name, request.tool_call["args"]):
8
            return ToolMessage(
9
                content="根据你的账户策略，当前工具调用被拦截。",
10
                tool_call_id=request.tool_call["id"],
11
            )
12
        return handler(request)

1
性能：额外抽象会增加序列化、检索、trace 和存储成本。
2
可靠性：可恢复机制要求状态 schema、幂等工具和错误分类清楚。
3
安全：metadata、state、tool args、retrieved docs 都可能包含敏感信息。
4
可观测性：至少记录 thread_id、node/tool、策略命中、错误和耗时。

1
公开 API 通常只负责归一化和装配；真正的运行语义在 runtime、协议对象和扩展点之间流转。

1
state、store、Document、metadata、trace、prompt messages 是不同协议；只有经过明确注入的内容才进入模型上下文。

1
可恢复异常可以转成 retry、fallback、interrupt；不可恢复异常必须上抛并进入 trace。

1
Public Config
2
  ↓
3
Normalize
4
  ↓
5
Build Runtime Object
6
  ↓
7
Attach Protocol / Extension / Storage

1
Input
2
  ↓
3
Runtime Entry
4
  ↓
5
Context / State / Config
6
  ↓
7
Core Execute
8
  ↓
9
State / Event / Output

1
正常路径 → 返回协议对象并更新必要状态
2
分支路径 → 路由、降级、拒答、人工确认或恢复
3
可恢复异常 → retry / fallback / repair / resume
4
不可恢复异常 → trace + raise
5
保护上限 → 停止循环、预算或高风险动作

1
第 16 篇：Agentic RAG 与 Retrieval Runtime 源码解剖

1
这项能力在旅行规划助手中应该放在哪一层？
2
它和前一篇能力如何协作？
3
哪些边界需要通过源码证据确认？