剖析 infini-agent-framework 的双层中间件:原理、实现与自定义

引言

在构建一个 Agent 应用时,我们几乎一定会遇到这样的需求:在每次 LLM 调用前后打日志、压缩过长的对话历史、给工具调用加一层错误兜底、给 HTTP 请求注入上下文、采集 Prometheus 指标……这些需求有一个共同特征——它们不是业务本身,而是横切在业务调用链路上的通用能力。把它们硬塞进业务代码会让系统迅速腐化,于是"中间件(Middleware)"这种模式应运而生。

infini-agent-framework(以下简称框架)作为一个基于 LangGraph + FastAPI 的 Python Agent 框架,在中间件的设计上做了一个值得细看的选择:它没有另起炉灶造一套自研中间件类型,而是直接复用了 Starlette 与 LangChain 两套成熟的中间件抽象,并在此基础上做了组合与编排。这导致框架内存在两套独立、互不共享类型的中间件系统:

层级 基类来源 作用范围 文件位置
HTTP 中间件 Starlette BaseHTTPMiddleware 每个 HTTP 请求 server/middleware/、server/extension/
Agent 中间件 LangChain AgentMiddleware 每次 Agent 运行(LLM 调用 / 工具调用 / 生命周期) langgraph/middlewares.py

本文从中间件的原理讲起,逐一拆解这两层的核心类型、链式执行机制、全部内置实现,最后给出三类自定义中间件的实战示例。

一、原理:洋葱模型与钩子链

1.1 什么是中间件

中间件的本质是对一个 callable 的包装(wrapper)。假设核心逻辑是一个函数 handler(request),中间件 M 把它包装成:

1
2
3
4
5
6
7
def M(request, handler):
    # 前置逻辑:在 handler 之前做点什么
    ...
    response = handler(request)      # 把控制权交给下一层
    # 后置逻辑:在 handler 之后做点什么
    ...
    return response

当多个中间件按顺序 M1 → M2 → M3 套在一起时,请求的执行路径就变成了一个"先逐层进入、再逐层退出"的嵌套结构,形如洋葱:

1
2
3
请求 ─▶ M1 前置 ─▶ M2 前置 ─▶ M3 前置 ─▶ 核心handler
                                                      ║
响应 ◀─ M1 后置 ◀─ M2 后置 ◀─ M3 后置 ◀───────────────╝

这种模型让每个中间件都能同时拥有"前"和"后"两个切入点,且彼此解耦——M2 不需要知道 M1 和 M3 的存在,它只关心自己手里的 handler。

1.2 框架为何选择"双层"

关键洞察是:HTTP 请求和 Agent 运行是两个粒度完全不同的执行单元。

  • 一个 HTTP 请求(如 POST /mission/run)可能触发一次完整的 Agent 运行,而一次 Agent 运行内部又可能发生数十次 LLM 调用和工具调用。
  • 如果只有 HTTP 层中间件,你想"在每次 LLM 调用前注入记忆"就无从下手——因为 HTTP 中间件根本感知不到单次 LLM 调用。
  • 反过来,Agent 层中间件也拿不到 HTTP 的请求头、状态码。

因此框架让 Starlette 负责"请求级"的横切(上下文、CORS、指标),让 LangChain AgentMiddleware 负责"调用级"的横切(提示词注入、历史压缩、工具兜底),各司其职。两者唯一的共同点只是"洋葱模型"这个思想。

二、HTTP 中间件层

2.1 核心类型

HTTP 中间件直接继承 Starlette 的 BaseHTTPMiddleware,只需实现一个 dispatch 方法:

1
2
3
4
5
6
7
8
from starlette.middleware.base import BaseHTTPMiddleware

class MyMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        # 前置
        response = await call_next(request)
        # 后置
        return response

注意签名里的 call_next 就是上面洋葱图里的 handler——调用它意味着"把请求交给下一层"。这套类型是 Starlette/FastAPI 原生的,框架并未重新定义。

2.2 链式执行机制:FastAPI 的 LIFO 栈

HTTP 中间件的"链"由 AgentServer._configure()(server/server.py:163)负责装配。它利用了 FastAPI add_middleware 的 LIFO(后进先出) 语义——最后添加的中间件最先执行(最内层)。装配顺序如下:

1
2
3
4
5
6
7
8
9
10
11
12
13
def _configure(self):
    self.__add_default_routes()
    self.__use_default_exception_handlers()
    # 1. 用户中间件:最先 add,最后执行(最外层)
    for middleware in reversed(self.middlewares):
        if middleware is not None:
            self.app.add_middleware(middleware)
    # 2. 扩展中间件:中间 add,中间执行
    for extension in self.default_extensions + self.user_extensions:
        extension.apply_to_app(self.app)
        extension.apply_middlewares(self.app)
    # 3. 默认中间件:最后 add,最先执行(最内层)
    self.__use_default_middlewares()

由此得到一条从内到外的执行链(越靠内越先执行、越接近业务路由):

1
2
CORSMiddleware  ──▶  ContextMiddleware / PrometheusMiddleware / MetricsMiddleware  ──▶  用户中间件  ──▶  路由handler
   (最内)                                                                                          (请求终点)

这个顺序的设计意图很清晰:默认的 CORS 必须最先执行(否则跨域预检会失败);扩展提供的 ContextMiddleware 需要在路由到达前把上下文准备好;而用户中间件放在最外层,可以看到完整的请求与响应。

2.3 内置 HTTP 中间件清单

中间件 文件 作用
ContextMiddleware server/middleware/context.py:46 调用 context_cls.from_request(request) 构建 DefaultGraphContext,挂到 request.state.context 上,供下游路由依赖注入使用
MetricsMiddleware server/middleware/metrics.py:87 为每个请求记录 Prometheus http_requests_total(Counter)与 http_request_duration_seconds(Histogram),按 method/endpoint/status 打标签
CORSMiddleware server/server.py:141 依据 config.http.cors 配置跨域头
PrometheusMiddleware server/extension/metrics.py:105 starlette-prometheus 默认指标导出

其中 ContextMiddleware 值得单独一提。它有一个配套的 get_context(request)(context.py:10)工具函数:即便该中间件尚未执行(例如被旁路调用),它也会惰性地创建并缓存一个默认 context,保证下游取上下文永不抛错:

1
2
3
4
5
6
7
8
9
10
11
def get_context(request: Request) -> DefaultGraphContext:
    context = getattr(request.state, "context", None)
    if context is not None:
        return context
    # 中间件还没跑,兜底创建一个默认 context 并缓存
    default_context = DefaultGraphContext(
        user_id=None, user_token=None, oidc_id_token=None, ...
        agent_manager_agent_name=get_config().name,
    )
    request.state.context = default_context
    return default_context

三、Agent 中间件层

3.1 核心类型与钩子

Agent 中间件继承自 LangChain 的 AgentMiddleware(langgraph/middlewares.py:7)。与 HTTP 中间件只有一个 dispatch 不同,AgentMiddleware 提供了一组细粒度的钩子,分别对应 Agent 运行生命周期的不同切点:

钩子 触发时机 典型用途
wrap_model_call / awrap_model_call 每次 LLM 调用前后 注入/改写提示词、打日志、链路追踪
wrap_tool_call / awrap_tool_call 每次工具调用前后 错误兜底、结果改写
before_agent / abefore_agent Agent 循环开始前 初始化状态
before_model / abefore_model LLM 调用前(状态级) 历史压缩、状态准备
after_model / aafter_model LLM 调用后(状态级) 抽取记忆、状态更新

其中 wrap_* 系列的签名是"包装器"形式,直接对应洋葱模型:

1
2
3
4
5
6
7
8
9
def wrap_model_call(
    self,
    request: ModelRequest,
    handler: Callable[[ModelRequest], ModelResponse],
) -> ModelCallResult:
    # 前置:可改写 request
    response = handler(request)      # 交给下一层 / 最终的 LLM 调用
    # 后置:可改写 response
    return response

而 before_* / after_* 系列是"状态钩子",返回 dict | None,返回的字典会被合并进图状态:

1
2
3
async def abefore_model(self, state, runtime) -> dict[str, Any] | None:
    ...
    return new_state   # 合并进 graph state

3.2 链式执行机制:委托给 create_agent

与 HTTP 层不同,Agent 层没有自定义的链装配器。框架直接把有序的中间件列表交给 LangChain 的 create_agent(..., middleware=<list>),由 LangGraph 引擎负责把 wrap_* 钩子嵌套成洋葱、把 before_*/after_* 钩子插成图节点。

装配发生在 DefaultPrebuiltReActAgent.__init__(prebuilt/agent.py:71),最终在 prebuilt/agent.py:141 处编译:

1
2
3
4
5
6
7
8
9
res = create_agent(
    model=...,
    tools=self.mcp_client.get_tools(),
    system_prompt=self.prompt_manager.get_prompt(),
    checkpointer=self.context_manager.checkpointer,
    state_schema=self.state_schema,
    context_schema=self.context_schema,
    middleware=self.middlewares,   # ← 链在这里织入
)

内置的装配顺序(agent.py:101-124)是一个精心设计的"先兜底、后增强"序列:

1
2
3
4
5
6
7
8
9
1. ToolErrorShieldMiddleware()         # 始终:工具异常兜底
2. ToolCallIdNormalizeMiddleware()    # 始终:规整 tool-call id
3. 用户传入的 middlewares              # 用户自定义(插入位)
4. LinkPromptMiddleware               # 若 langfuse 启用
5. MultiModelRagMiddleware           # 若 rag 启用
   DuplicateToolCallMiddleware
6. CompressionMiddleware              # 若 compression 启用
7. MemoryMiddleware                  # 若 memory 启用
8. SkillsMiddleware                  # 若 skills 启用

注意第 3 步:用户中间件被插在两个"始终开启的防护盾"之后、所有配置条件型中间件之前。这意味着用户的 wrap_model_call 看到的是已经被规整过 tool-call id 的模型响应,但它注入的内容会在 RAG / 记忆 / 技能等增强之前生效。这个位置对大多数场景是合理的。

3.3 内置 Agent 中间件清单

中间件 钩子 作用
LinkPromptMiddleware (middlewares.py:48) wrap_model_call 把当前 Langfuse generation 与 prompt 管理器的 prompt client 关联,用于可观测性
CompressionMiddleware (middlewares.py:76) abefore_model 在 LLM 调用前通过 Compressor 压缩过长的对话历史
PrettyPrintMiddleware (middlewares.py:132) wrap_model_call 以美化格式打印送入模型的 messages,调试用(测试中常用)
MultiModelRagMiddleware (middlewares.py:153) wrap_tool_call 从 RAG 工具返回的 ToolMessage 中抽取图片内容,支持多模态
MemoryMiddleware (middlewares.py:201) wrap_model_call(注入)+ aafter_model(抽取)+ tools 调用前把检索到的记忆作为 SystemMessage 注入;调用后从对话中异步抽取新记忆。支持 MESSAGE/TOOL/AUTO_EXTRACT_AND_TOOL 三种交互模式
DuplicateToolCallMiddleware (middlewares.py:350) wrap_model_call 检测自上一条 HumanMessage 以来重复的工具调用/doc-id,超过 max_duplicate_searches 时注入一条"停止搜索"的 HumanMessage
SkillsMiddleware (middlewares.py:451) wrap_model_call 把 skills_manager.prompt 追加到系统提示词末尾
ToolErrorShieldMiddleware (middlewares.py:472) wrap_tool_call 捕获工具异常,返回 ToolMessage(status="error"),避免异常冒出 graph 流终止 SSE;asyncio.CancelledError 与 GraphBubbleUp 仍会 re-raise
ToolCallIdNormalizeMiddleware (middlewares.py:530) wrap_model_call 为空/重复的 tool-call id 分配新的 lc_<uuid>,保证 ToolMessage 与 tool-call 能正确对应
FormatToolCallMiddleware (middlewares.py:585) before_agent + wrap_tool_call 把工具返回的 Document 加工成带 doc_id/chunk_id 引用指令的参考材料 JSON,并对已见观点去重
ConditionalHITLMiddleware (middlewares.py:802) after_model 继承 HumanInTheLoopMiddleware,支持按工具调用的谓词函数(ToolCallPredicate)条件触发人工介入

一个值得细读的实现:ToolErrorShieldMiddleware

它是理解"工具包装器 + 兜底"模式的最佳样本。注释写得很清楚:没有这个盾,未捕获的工具异常会冒出 graph 流、以 FAIL 终止 SSE 响应并结束整个 mission;有了它,LLM 看到的是一条 status="error" 的 ToolMessage,可以自行决定重试、换工具还是优雅放弃:

1
2
3
4
5
6
7
8
9
10
11
class ToolErrorShieldMiddleware(AgentMiddleware):
    async def awrap_tool_call(
        self, request: ToolCallRequest,
        handler: Callable[[ToolCallRequest], Awaitable[ToolMessage | Command]],
    ) -> ToolMessage | Command:
        try:
            return await handler(request)
        except (asyncio.CancelledError, GraphBubbleUp):
            raise                          # 用户主动 pause/cancel、HITL 中断,放行
        except BaseException as e:
            return self._build_error_message(request, e)

注意它只对 asyncio.CancelledError 和 GraphBubbleUp 放行——前者是用户主动取消,后者是 human-in-the-loop 的中断机制,这两类异常必须穿透中间件才能让上层控制流正常工作。这种"宽进严出"的异常处理是写健壮中间件的范式。

注入型中间件:SkillsMiddleware

它展示了"改写请求"的最小写法——不调用 handler 之前先把 request.override(...) 出来的新请求传下去:

1
2
3
4
5
6
def _override_system_prompt(self, request: ModelRequest) -> ModelRequest:
    system_prompt = (request.system_prompt or "") + self.skills_manager.prompt
    return request.override(system_prompt=system_prompt)

async def awrap_model_call(self, request, handler) -> ModelCallResult:
    return await handler(self._override_system_prompt(request))

ModelRequest.override(...) 会返回一个新的不可变请求副本,这是框架/LangChain 提供的关键原语——它保证了中间件不会意外篡改上游看到的原始请求,每一层都拿到自己的快照。

3.4 深入:中间件在运行时究竟是怎么被调用起来的

读完前面的注册逻辑,一个自然的疑问是:既然 self.middlewares 只是一个列表、create_agent(middleware=...) 把它收下,那 Agent 跑到某个 stage(比如工具调用)时,是谁、在什么时机、怎么把列表里那些中间件的 wrap_tool_call 串起来的?是每次跑到该 stage 都临时遍历一遍列表、逐个检查它有没有重写对应 hook 吗?

答案是否定的——"检查某个中间件是否重写了对应 hook"这件事,确实存在,但它发生在编译期(create_agent 调用时),不是运行时临场检查。编译期把每个 hook 点的重写者静态织成一条嵌套链,运行时直接调用这条预织好的链。

下面拆开 LangChain 源码(langchain/agents/factory.py)讲清这三步。

(1)基类的默认实现:要么空、要么报错

AgentMiddleware 基类把所有 hook 都定义了出来(types.py:192),但默认实现分两类:

  • wrap_model_call / awrap_model_call / wrap_tool_call / awrap_tool_call 的默认实现是 raise NotImplementedError(见 types.py:308、types.py:431)——所以基类不会"静默放行",而是显式报错。
  • before_agent / before_model / after_model / after_agent 的默认实现是 返回 None(空操作,等同不介入)。

这个差异决定了第二步的过滤为何必要:如果把一个只重写了 wrap_tool_call 的中间件(比如 ToolErrorShieldMiddleware)误塞进 model call 链,它继承的基类 wrap_model_call 会直接抛 NotImplementedError。因此编译期必须、也确实做了"按 hook 点过滤重写者"的动作。

(2)编译期过滤:用方法对象的身份比对判断"是否重写"

create_agent 对每个 hook 点分别过滤出一组中间件,判断方式很巧妙——不靠 hasattr(因为基类已经定义了所有 hook,hasattr 恒为真),而是靠方法对象的身份比对 is not(factory.py:760):

1
2
3
4
5
6
# Collect middleware with wrap_model_call or awrap_model_call hooks
middleware_w_wrap_model_call = [
    m for m in middleware
    if m.__class__.wrap_model_call is not AgentMiddleware.wrap_model_call
    or m.__class__.awrap_model_call is not AgentMiddleware.awrap_model_call
]

m.__class__.wrap_model_call is not AgentMiddleware.wrap_model_call 的含义是:这个子类上的 wrap_model_call,还是不是基类那个未被重写的函数对象。

  • 若子类重写了它,m.__class__.wrap_model_call 指向新的函数对象 → is not 成立 → 被选中;
  • 若没重写,它和基类方法是同一个对象 → is not 不成立 → 被排除。

注释里还点明了一个细节:只要中间件重写了 wrap_model_call 或 awrap_model_call 中的任意一个,它就会同时进入"同步链"和"异步链"两个过滤结果——“Include middleware with either implementation to ensure NotImplementedError is raised when middleware doesn’t support the execution path”。也就是说,如果你只实现了 sync 版本却以 async 方式运行 agent,那条链里会保留它、调用时自然抛 NotImplementedError,把"用错执行模式"的问题显式暴露给你,而不是悄悄跳过。

(3)编译期织链:重写者按"first = outermost"折叠成一条嵌套链

过滤完不是把列表存起来等运行时遍历,而是立刻折叠(fold)成一条嵌套的可调用对象(factory.py:779、_chain_tool_call_wrappers factory.py:419):

1
2
sync_handlers = [m.wrap_model_call for m in middleware_w_wrap_model_call]
wrap_model_call_handler = _chain_model_call_handlers(sync_handlers)

折叠逻辑(compose_two)把 wrappers[:-1] 反向(reversed) 依次套到最内层 wrappers[-1] 上,结果是列表里第一个中间件变成最外层。源码注释白纸黑字写明了这条规则:

factory.py:82:Composes handlers so first in list becomes outermost layer。
factory.py:400:Compose wrappers into middleware stack (first = outermost)。

请求流向也举例得很直白:_chain_tool_call_wrappers([auth, cache, retry]) → auth -> cache -> retry -> tool,响应反向回流。

这恰好印证了 ToolErrorShieldMiddleware 被放在 [ToolErrorShieldMiddleware(), ToolCallIdNormalizeMiddleware(), ...] 链首的设计意图——它最外层,能兜住所有内层中间件与真实工具执行抛出的异常。

(4)两类 hook 接入 graph 的方式不同

编译期还要区分 hook 的形态,因为它们接入 graph 的方式不一样:

hook 类别 接入方式 运行时
wrap_model_call / wrap_tool_call 折叠成一个嵌套可调用对象,包在真实 model/tool 执行外层(洋葱) 每次 model/tool 调用都走这条预织链
before_agent / before_model / after_model / after_agent 同样按身份比对过滤,但折叠后挂在 graph 的独立节点上(见 factory.py:755 对 aafter_agent 的同款判断) 节点执行时调用,返回的 dict 合并进图状态

这就是 CompressionMiddleware 的 abefore_model 与 ToolErrorShieldMiddleware 的 awrap_tool_call 的本质区别:前者是图节点,产物是状态更新;后者是洋葱层,产物是对工具调用的包装。两者机制不同,但"按 hook 点过滤重写者"这一步是共用的。

小结:校准后的调用链路

把上面三步串起来,运行时调用链路其实是:

  1. 编译期:create_agent(middleware=[...]) 对每个 hook 点,用 m.__class__.<hook> is not AgentMiddleware.<hook> 过滤出"重写了该 hook"的中间件子集;
  2. 编译期:把每个子集按"first = outermost"折叠成一条嵌套可调用对象(wrap_*)或一个图节点(before_*/after_*),写进编译后的 graph;
  3. 运行时:Agent 跑到某个 stage 时不再扫描 middleware 列表,直接调用编译期织好的那条链/那个节点。某层中间件是否被调用,在编译期就已静态决定。

所以——"检查是否重写"是对的,但它是一次性的编译期动作,产物是一条预织好的链;运行时只是调用这条链,没有动态检查、没有逐个遍历的开销。这也是为什么 create_agent 在 agent 启动时执行的 self.logger.info(f"graph ascii: \n{res.get_graph().draw_ascii()}")(agent.py:153) 能直接画出一张包含各 hook 节点的拓扑图——链和节点在那个时刻就已经织好了。

四、如何自定义中间件

理解了原理和内置实现,自定义就只剩"选层、选钩子、填逻辑"三步。下面给出三个实战示例,覆盖最常见的诉求。

示例 1:Agent 层——给每次 LLM 调用加敏感词过滤

这是一个典型的 wrap_model_call 包装器,前置过滤输入、后置过滤输出:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
from collections.abc import Awaitable, Callable
from langchain.agents.middleware import AgentMiddleware, ModelRequest, ModelResponse
from langchain.agents.middleware.types import ModelCallResult

SENSITIVE_WORDS = {"password", "secret"}

class SensitiveFilterMiddleware(AgentMiddleware):
    def __init__(self) -> None:
        pass

    def _redact(self, text: str) -> str:
        for w in SENSITIVE_WORDS:
            text = text.replace(w, "***")
        return text

    def wrap_model_call(
        self,
        request: ModelRequest,
        handler: Callable[[ModelRequest], ModelResponse],
    ) -> ModelCallResult:
        # 前置:可选,对入参做处理
        response = handler(request)
        # 后置:对模型输出做脱敏
        if hasattr(response, "content") and isinstance(response.content, str):
            response.content = self._redact(response.content)
        return response

    async def awrap_model_call(
        self,
        request: ModelRequest,
        handler: Callable[[ModelRequest], Awaitable[ModelResponse]],
    ) -> ModelCallResult:
        response = await handler(request)
        if hasattr(response, "content") and isinstance(response.content, str):
            response.content = self._redact(response.content)
        return response

注册到预构建 Agent,只需在构造时传入:

1
2
3
4
5
6
7
8
from infini_agent_framework.prebuilt import DefaultPrebuiltReActAgent
from infini_agent_framework.server import AgentServer, DefaultAgenticAppExtension

agent = await DefaultPrebuiltReActAgent(
    middlewares=[SensitiveFilterMiddleware()],
).must_initialize()
server = AgentServer(DefaultAgenticAppExtension(agent))
server.run()

它会被插在内置顺序的第 3 位——在两个防护盾之后、配置条件型增强之前,正合适。

示例 2:HTTP 层——统一请求鉴权

HTTP 中间件继承 BaseHTTPMiddleware,通过 AgentServer(middlewares=...) 注册。注意 middlewares 接收的是类(不是实例):

1
2
3
4
5
6
7
8
9
10
11
12
13
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import JSONResponse
from collections.abc import Callable

VALID_TOKEN = "Bearer my-secret"

class AuthMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next: Callable):
        auth = request.headers.get("Authorization")
        if auth != VALID_TOKEN:
            return JSONResponse({"detail": "unauthorized"}, status_code=401)
        return await call_next(request)
1
2
3
4
server = AgentServer(
    DefaultAgenticAppExtension(agent),
    middlewares=[AuthMiddleware],
)

由于 LIFO 语义,用户中间件执行在最外层——它在 ContextMiddleware、MetricsMiddleware 之前先看到请求,因此这里拦截 401 不会触发后续的指标/上下文逻辑(取决于你是否希望如此;若想让 401 也被指标统计,则需要把它放到更内层,见示例 3 的扩展方式)。

示例 3:通过扩展(Extension)注册中间件

当中间件需要依赖注入、需要配置、或需要控制执行顺序时,更优雅的方式是写一个 AppExtension。AppExtension(server/extension/base.py:52)是一个抽象基类,只需实现 name、apply_to_app、health_handler 三个方法,可选覆盖 apply_middlewares:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
from fastapi import FastAPI, Request
from starlette.middleware.base import BaseHTTPMiddleware
from collections.abc import Callable
from infini_agent_framework.server.extension.base import AppExtension

class RateLimitMiddleware(BaseHTTPMiddleware):
    def __init__(self, app, limit: int = 100):
        super().__init__(app)
        self.limit = limit
        self._counter: dict[str, int] = {}

    async def dispatch(self, request: Request, call_next: Callable):
        client = request.client.host if request.client else "unknown"
        self._counter[client] = self._counter.get(client, 0) + 1
        if self._counter[client] > self.limit:
            from starlette.responses import JSONResponse
            return JSONResponse({"detail": "rate limited"}, status_code=429)
        return await call_next(request)

class RateLimitExtension(AppExtension):
    def __init__(self, limit: int = 100):
        self.limit = limit

    @property
    def name(self) -> str:
        return "rate_limit"

    def apply_middlewares(self, app: FastAPI) -> None:
        app.add_middleware(RateLimitMiddleware, limit=self.limit)

    def apply_to_app(self, app: FastAPI) -> None:
        pass  # 不注册额外路由

    async def health_handler(self, request: Request) -> bool:
        return True
1
2
3
4
server = AgentServer(
    DefaultAgenticAppExtension(agent),
    RateLimitExtension(limit=60),
)

扩展中间件执行在中间层(默认中间件之后、用户中间件之前),正好让限流能享受到 CORS 但仍在上下文注入之前拦截,顺序可控。框架内置的 MetricsExtension 与 DefaultAgenticAppExtension 就是这么把自己的 PrometheusMiddleware/MetricsMiddleware/ContextMiddleware 注册进去的。

五、设计要点回顾

把全文收束为几条可迁移的设计经验:

  1. 按执行粒度分层。请求级横切用 HTTP 中间件,调用级横切用 Agent 中间件,不要让一层勉强承担另一层的职责。
  2. 复用成熟抽象。框架没有自造中间件类型,而是复用 Starlette 与 LangChain 的基类,既减少学习成本,也避免了与生态割裂。
  3. 链顺序是设计决策。HTTP 层用 LIFO 把默认中间件放最内、用户中间件放最外;Agent 层用"先兜底(Shield)、后规整(Normalize)、再增强(RAG/Memory/Skills)"的固定序。用户自定义时要意识到自己被插在哪里。
  4. 重写检查发生在编译期。Agent 中间件的"某个中间件要不要在某个 hook 点生效"是 create_agent 时用方法对象身份比对(is not)静态判定的,并立刻折叠成一条预织链(first = outermost);运行时不做动态扫描。意味着:只重写 wrap_tool_call 的中间件不会误入 model call 链,但若你只写了 sync 版却在 async 上下文运行,会显式抛 NotImplementedError 而非悄悄跳过。
  5. 请求不可变。用 request.override(...) 改写而不是原地修改,让洋葱的每一层都拿到隔离的快照。
  6. 异常要分层处理。ToolErrorShieldMiddleware 示范了"兜底一切,但放行 CancelledError/GraphBubbleUp"——哪些异常该吞、哪些该穿透,是写健壮中间件的核心判断。

中间件是 Agent 框架里最优雅的扩展点之一:它把横切能力从业务里剥离,让 Agent 的主循环保持干净,同时让每一项通用能力都可插拔、可组合、可替换。理解了这套双层洋葱,你就掌握了给 infini-agent-framework 做任何"非侵入式增强"的钥匙。

本文基于 infini-agent-framework 源码梳理,关键文件:langgraph/middlewares.py(全部 11 个 Agent 中间件)、prebuilt/agent.py(Agent 中间件装配)、server/server.py(HTTP 中间件 LIFO 链)、server/middleware/ 与 server/extension/(HTTP 中间件实现)。