s02
Agent 架构与中间件链
架构层14 层中间件的编排与定位机制
@Next/@Prev 装饰器精确控制中间件位置每层中间件解决一个生产痛点，组合出工业级 Context Engineering
14 层中间件的编排与定位机制
@Next/@Prev 装饰器精确控制中间件位置每层中间件解决一个生产痛点，组合出工业级 Context Engineering
裸 LLM 调用只是一问一答。生产环境中 Agent 需要处理：上下文过长时自动摘要、死循环检测、工具调用异常恢复、子任务并发限制、内存更新、标题生成、Token 计费……每个横切关注点独立成中间件，通过 before_model/after_model 钩子精确插入处理逻辑，避免 prompt 膨胀和 Agent 逻辑污染。
工具调用流式响应
人工反馈中断点
工具重试与错误恢复
最大步数限制
自动确认机制
子 Agent 编排
长期记忆读写
技能系统注入
沙箱环境管理
模型动态切换
人格系统注入
上下文裁剪
提示词模板
日志记录
中间件链组装 — 严格的顺序保证：Summarization 最早缩上下文，Clarification 最后拦截
def _build_middlewares(config, model_name, agent_name=None):
middlewares = build_lead_runtime_middlewares(lazy_init=True)
# Summarization: 最早执行, 缩减上下文
if summarization_middleware := _create_summarization_middleware():
middlewares.append(summarization_middleware)
middlewares.append(TokenUsageMiddleware()) # 计费
middlewares.append(TitleMiddleware()) # 标题生成
middlewares.append(MemoryMiddleware()) # 记忆更新
if model_config.supports_vision:
middlewares.append(ViewImageMiddleware()) # 图片注入
middlewares.append(SubagentLimitMiddleware()) # 并发限制
middlewares.append(LoopDetectionMiddleware()) # 死循环检测
middlewares.append(ClarificationMiddleware()) # 最后: 澄清拦截
return middlewaresSubagentLimitMiddleware — 硬性截断超额的并行 task 调用，比 prompt 约束更可靠
class SubagentLimitMiddleware(AgentMiddleware):
def __init__(self, max_concurrent: int = 3):
self.max_concurrent = _clamp_subagent_limit(max_concurrent)
def after_model(self, state, runtime):
tool_calls = last_msg.tool_calls
task_indices = [i for i, tc in enumerate(tool_calls)
if tc.get("name") == "task"]
if len(task_indices) <= self.max_concurrent:
return None
# 截断超额调用
indices_to_drop = set(task_indices[self.max_concurrent:])
truncated = [tc for i, tc in enumerate(tool_calls)
if i not in indices_to_drop]
return {"messages": [msg.model_copy(
update={"tool_calls": truncated})]}