EgoAlpha6.几种中间件(Middleware)的详细解释
中间件（Middleware）是 LangChain v1 引入的可插拔扩展点。中间件可以在代理（Agent）执行的每一步前后插入自定义逻辑：修改模型请求、注入状态 schema、注册工具、拦截/改写工具调用、合并状态等。 你可以定义任意数量的中间件，并使用他们来创建agent。
deepagent_middleware = [
TodoListMiddleware(),
FilesystemMiddleware(backend=backend),
SubAgentMiddleware(
default_model=model,
default_tools=tools,
subagents=subagents if subagents is not None else [],
default_middleware=[
TodoListMiddleware(),
FilesystemMiddleware(backend=backend),
SummarizationMiddleware(
model=model,
max_tokens_before_summary=170000,
messages_to_keep=6,
),
AnthropicPromptCachingMiddleware(unsupported_model_behavior="ignore"),
PatchToolCallsMiddleware(),
],
default_interrupt_on=interrupt_on,
general_purpose_agent=True,
),
SummarizationMiddleware(
model=model,
max_tokens_before_summary=170000,
messages_to_keep=6,
),
AnthropicPromptCachingMiddleware(unsupported_model_behavior="ignore"),
PatchToolCallsMiddleware(),
]
以上列出的中间件是系统预定义的几种基础模块，分别赋予智能体以下核心能力：
- 规划能力（TodoList） —— 使代理能够生成并管理任务计划；
- 文件系统交互能力（Filesystem） —— 允许代理访问、读取、编辑和写入文件；
- 子代理管理能力（SubAgent） —— 使代理能够根据需要创建并调用独立的子代理以处理特定子任务。
中间件是可堆叠且可扩展的，这意味着你可以根据需要向同一个代理中添加多个中间件。系统会按照它们被添加的顺序依次应用这些中间件，从而逐步增强代理的功能和行为模式。
在接下来的章节中，我们将逐一介绍这些中间件的具体作用与内部实现机制。
6.1 TodoListMiddleware
class PlanningMiddleware(AgentMiddleware):
state_schema = PlanningState
tools = [write_todos]
def modify_model_request(self, request: ModelRequest, agent_state: PlanningState) -> ModelRequest:
request.system_prompt = request.system_prompt + "\n\n" + WRITE_TODOS_SYSTEM_PROMPT
return request
#这是原PlanningMiddleware的代码，也就是TodoListMiddleware改名之前的叫法
规划中间件 的核心作用在于扩展智能体的运行机制，包括以下三方面：
- 扩展代理的状态模式 中间件首先可以修改或扩展代理的状态结构。例如，规划中间件在原有状态（
AgentState）的基础上新增了一个todos字段，用于存储代理的待办任务列表（To-Do List）。通过这种方式，代理能够在执行过程中持续追踪计划、更新任务状态，从而具备更强的长期规划与任务管理能力。 - 扩展工具的使用范围与数量 除了扩展状态，中间件还可以向代理注册额外的工具（Tools），或限制代理在特定阶段只能使用某些工具。 在规划中间件中，我们仅注册了一个“写入工具”（
write_todos），它的作用是向状态中写入新的待办事项，从而动态更新规划状态。 - 修改模型请求并注入自定义提示 中间件还可以通过实现
modify_model_request()方法来调整模型请求（Model Request）。在规划中间件中，该方法会在系统提示（System Prompt）的末尾追加一段自定义说明，用于指导模型如何使用已注册的工具（如write_todos）来更新待办任务列表。这一机制确保模型能够理解当前可用的工具及其用法，从而实现更稳定、更具可控性的行为规划。
TodoListMiddleware 为您的代理提供了一个专门用于更新此待办事项列表的工具。在执行多部分任务之前和执行时，系统会提示代理使用 write_todos 工具来跟踪其执行的动作以及仍需要执行的动作。
6.2 FilesystemMiddleware
class FilesystemMiddleware(AgentMiddleware):
state_schema = FilesystemState
tools = [ls, read_file, write_file, edit_file]
def modify_model_request(self, request: ModelRequest, agent_state: FilesystemState) -> ModelRequest:
request.system_prompt = request.system_prompt + "\n\n" + FILESYSTEM_SYSTEM_PROMPT
return request
文件系统（Filesystem）中间件 允许智能体在状态（State）中维护一个持久化的文件系统结构，从而扩展了原有的状态模型。在该机制下，状态中会包含一个名为 files 的键，其对应的值是一个字典，用于存储文件名与文件内容的映射关系。
借助文件系统中间件，智能体可以使用多种工具来操作这些文件，包括：
- ls：列出文件系统中的文件
- read_file：从文件中读取整个文件或一定行数
- write_file：将新文件写入文件系统
- edit_file：编辑文件系统中的现有文件
此外，文件系统中间件还可以通过修改模型请求（modify_model_request）来增强模型的上下文提示（prompt）。该机制会自动在系统提示中添加说明，指导模型如何使用文件操作工具以更新本地文件系统状态，从而实现更持久、可追踪的任务执行过程。
6.3 SubAgentMiddleware
class SubAgentMiddleware(AgentMiddleware):
def __init__(
self,
*,
default_model: str | BaseChatModel,
default_tools: Sequence[BaseTool | Callable | dict[str, Any]] | None = None,
default_middleware: list[AgentMiddleware] | None = None,
default_interrupt_on: dict[str, bool | InterruptOnConfig] | None = None,
subagents: list[SubAgent | CompiledSubAgent] | None = None,
system_prompt: str | None = TASK_SYSTEM_PROMPT,
general_purpose_agent: bool = True,
task_description: str | None = None,
) -> None:
"""Initialize the SubAgentMiddleware."""
super().__init__()
self.system_prompt = system_prompt
task_tool = _create_task_tool(
default_model=default_model,
default_tools=default_tools or [],
default_middleware=default_middleware,
default_interrupt_on=default_interrupt_on,
subagents=subagents or [],
general_purpose_agent=general_purpose_agent,
task_description=task_description,
)
self.tools = [task_tool]
def wrap_model_call(
self,
request: ModelRequest,
handler: Callable[[ModelRequest], ModelResponse],
) -> ModelResponse:
"""Update the system prompt to include instructions on using subagents."""
if self.system_prompt is not None:
request.system_prompt = request.system_prompt + "\n\n" + self.system_prompt if request.system_prompt else self.system_prompt
return handler(request)
async def awrap_model_call(
self,
request: ModelRequest,
handler: Callable[[ModelRequest], Awaitable[ModelResponse]],
) -> ModelResponse:
"""(async) Update the system prompt to include instructions on using subagents."""
if self.system_prompt is not None:
request.system_prompt = request.system_prompt + "\n\n" + self.system_prompt if request.system_prompt else self.system_prompt
return await handler(request)
SubAgentMiddleware 是子代理中间件，用于向主 agent 注入一个 task 工具（task tool），该工具可以用来临时启动子代理（subagent），把复杂/上下文重的子任务交给子代理在隔离的上下文中执行，然后把子代理的最终结果汇回主线程。
6.3.1职责
- 向主 Agent 注入一个
task工具，使主 Agent 能够动态选择并调用子代理（SubAgent或CompiledSubAgent）。 - 注册多个可用的子代理实例，每个子代理可拥有独立的模型、工具集与中间件栈，从而实现职责分离与上下文隔离。
为何以中间件形式实现子代理而非普通工具？
模块化扩展性**：子代理不仅需要定义额外的系统提示（prompt）、状态 schema 与工具集，还可能使用独立的模型或中间件栈。将其设计为中间件，可以在不修改主 Agent 架构的前提下，实现完整的子代理能力注入。**上下文控制能力：中间件可以在系统提示中动态插入指导信息（例如何时、如何调用子代理），从而使主 Agent 在决策阶段具备更高层次的调度与调用控制。
6.3.2_create_task_tool
"""Create a task tool for invoking subagents.
Args:
default_model: Default model for subagents.
default_tools: Default tools for subagents.
default_middleware: Middleware to apply to all subagents.
default_interrupt_on: The tool configs to use for the default general-purpose subagent. These
are also the fallback for any subagents that don't specify their own tool configs.
subagents: List of subagent specifications.
general_purpose_agent: Whether to include general-purpose agent.
task_description: Custom description for the task tool. If `None`,
uses default template. Supports `{available_agents}` placeholder.
Returns:
A StructuredTool that can invoke subagents by type.
"""
_create_task_tool 的作用是创建一个名为 "task" 的工具，使主 Agent 可以调用不同类型的子代理。它首先通过 _get_subagents() 获取可用子代理及其描述，然后根据传入的 subagent_type 调用对应子代理，以独立上下文执行任务。执行结果会被封装成 Command 更新主 Agent 的状态。最终，函数返回一个 StructuredTool 对象，使主 Agent 能通过 task() 动态委派任务。
6.3.3_get_subagents
"""Create subagent instances from specifications.
Args:
default_model: Default model for subagents that don't specify one.
default_tools: Default tools for subagents that don't specify tools.
default_middleware: Middleware to apply to all subagents. If `None`,
no default middleware is applied.
default_interrupt_on: The tool configs to use for the default general-purpose subagent. These
are also the fallback for any subagents that don't specify their own tool configs.
subagents: List of agent specifications or pre-compiled agents.
general_purpose_agent: Whether to include a general-purpose subagent.
Returns:
Tuple of (agent_dict, description_list) where agent_dict maps agent names
to runnable instances and description_list contains formatted descriptions.
"""
_get_subagents 用于根据默认配置和用户自定义设置生成可运行的子代理集合，并返回每个子代理的描述文本。它会创建通用子代理（如启用）并处理自定义子代理的模型、工具和中间件。返回的子代理字典和描述列表会被 _create_task_tool 使用，用于构建 "task" 工具，使主 Agent 能动态调用不同类型的子代理完成任务。
总体来说，子代理中间件的主要职责是：
- 构建并维护子代理列表；
- 通过“任务工具（Task Tool）”公开这些子代理，使主代理能够根据任务需求动态调用它们；
- 当主代理决定启动任务时，会通过任务工具选择要调用的子代理，并传递必要的上下文与任务信息；
- 随后，子代理会独立执行任务，最终将结构化、完整的结果返回给主代理进行整合。
这种机制使得 DeepAgent 能够像一个多层级协作系统一样运作，主代理负责总体规划与任务分配，而子代理则承担具体执行，从而实现真正意义上的分层自治与并行智能。
6.4PatchToolCallsMiddleware
class PatchToolCallsMiddleware(AgentMiddleware):
"""Middleware to patch dangling tool calls in the messages history."""
def before_agent(self, state: AgentState, runtime: Runtime[Any]) -> dict[str, Any] | None: # noqa: ARG002
"""Before the agent runs, handle dangling tool calls from any AIMessage."""
messages = state["messages"]
if not messages or len(messages) == 0:
return None
patched_messages = []
# Iterate over the messages and add any dangling tool calls
for i, msg in enumerate(messages):
patched_messages.append(msg)
if msg.type == "ai" and msg.tool_calls:
for tool_call in msg.tool_calls:
corresponding_tool_msg = next(
(msg for msg in messages[i:] if msg.type == "tool" and msg.tool_call_id == tool_call["id"]),
None,
)
if corresponding_tool_msg is None:
# We have a dangling tool call which needs a ToolMessage
tool_msg = (
f"Tool call {tool_call['name']} with id {tool_call['id']} was "
"cancelled - another message came in before it could be completed."
)
patched_messages.append(
ToolMessage(
content=tool_msg,
name=tool_call["name"],
tool_call_id=tool_call["id"],
)
)
return {"messages": [RemoveMessage(id=REMOVE_ALL_MESSAGES), *patched_messages]}
PatchToolCallsMiddleware在 agent 执行前扫描消息历史，把那些“悬挂的工具调用”（dangling tool calls）修复成明确的 ToolMessage。悬挂工具调用指 AI 消息里有 tool_calls，但在消息历史中没有对应的 ToolMessage（可能被打断、取消或丢失）。
工作方式：
- 遍历消息历史，检查 AI 消息的工具调用。
- 对没有对应
ToolMessage的工具调用生成占位消息。 - 用
RemoveMessage清空旧消息，再写入修复后的消息。
6.5其他中间件说明
SummarizationMiddleware：
当对话历史或上下文变得非常长（超过设定阈值）时，该中间件会自动对早期消息做压缩总结，保留 messages_to_keep 条最近消息与必要摘要，避免超出上下文窗口，同时保留关键信息。
HumanInTheLoopMiddleware:
在需要人工审查工具调用或关键决策时，该中间件可以暂停自动流程，把决策点暴露给人类审阅（适用于高风险或需要合规审查的场景）。