05-Agent 智能体开发实战指南（五）：中间件系统与动态提示词

Agent 智能体开发实战指南（五）：中间件系统与动态提示词

系列导读：这是《Agent 智能体开发实战指南》系列的第五篇，将深入讲解 LangChain Agent 的中间件系统，包括 Hooks 机制、工具调用监控、动态提示词切换等高级功能，让你能够完全掌控 Agent 的执行流程。

一、什么是中间件？

1.1 核心概念

中间件（Middleware）：在 Agent 执行过程中拦截和修改行为的组件。

类比理解：

用户请求 → [中间件 1] → [中间件 2] → [Agent] → [中间件 3] → 响应 ↓ ↓ ↓ ↓ 日志记录 权限检查 执行逻辑 结果格式化

1.2 为什么需要中间件？

1.3 中间件的价值

• 解耦：业务逻辑与横切关注点分离
• 复用：一个中间件可用于多个 Agent
• 灵活：运行时动态添加/移除
• 可观测：统一监控和调试

二、Hooks 机制详解

2.1 六类钩子函数

LangChain 提供两类钩子：

节点式钩子（执行点顺序拦截）：

包装式钩子（针对工具和模型）：

2.2 钩子执行顺序

用户请求 ↓ ┌─────────────────┐ │ before_agent │ ← 第 1 步：Agent 执行前 └────────┬────────┘ ↓ ┌─────────────────┐ │ before_model │ ← 第 2 步：模型调用前 └────────┬────────┘ ↓ ┌─────────────────┐ │ wrap_model_call│ ← 第 3 步：包装模型调用 └────────┬────────┘ ↓ ┌─────────────────┐ │ Model 执行 │ └────────┬────────┘ ↓ ┌─────────────────┐ │ after_model │ ← 第 4 步：模型调用后 └────────┬────────┘ ↓ ┌─────────────────┐ │ wrap_tool_call │ ← 第 5 步：工具调用（如有） └────────┬────────┘ ↓ ┌─────────────────┐ │ after_agent │ ← 第 6 步：Agent 执行后 └────────┬────────┘ ↓ 返回响应

三、实战：工具调用监控

3.1 使用 @wrap_tool_call

fromtypingimportCallablefromlangchain.agents.middlewareimportwrap_tool_callfromlangchain.tools.tool_nodeimportToolCallRequestfromlangchain_core.messagesimportToolMessagefromutils.logger_handlerimportlogger@wrap_tool_calldefmonitor_tool(request:ToolCallRequest,# 请求数据封装handler:Callable[[ToolCallRequest],ToolMessage],# 执行的函数本身)->ToolMessage:"""工具执行监控中间件"""# 执行前：记录日志logger.info(f"[tool monitor] 执行工具：{request.tool_call['name']}")logger.info(f"[tool monitor] 传入参数：{request.tool_call['args']}")try:# 执行工具result=handler(request)# 执行后：记录成功日志logger.info(f"[tool monitor] 工具{request.tool_call['name']}调用成功")# 特殊处理：检测到报告生成工具，标记上下文ifrequest.tool_call['name']=="fill_context_for_report":request.runtime.context["report"]=Truelogger.info("[tool monitor] 检测到报告生成场景，已标记上下文")returnresultexceptExceptionase:# 异常处理logger.error(f"工具{request.tool_call['name']}调用失败，原因：{str(e)}")raisee

3.2 关键参数解析

ToolCallRequest 结构：

ToolCallRequest(tool_call={"name":"get_weather",# 工具名"args":{"city":"深圳"},# 参数"id":"call_abc123"# 调用 ID},runtime=Runtime(context={"report":False}# 运行时上下文))

handler 函数：

• 实际执行工具的函数
• 必须调用它，否则工具不会执行
• 可以修改参数或结果

3.3 应用场景

1. 日志记录：

@wrap_tool_calldeflog_tool_call(request,handler):start_time=time.time()result=handler(request)duration=time.time()-start_time logger.info(f"工具{request.tool_call['name']}耗时：{duration:.2f}s")returnresult

2. 重试机制：

@wrap_tool_calldefretry_tool_call(request,handler):max_retries=3foriinrange(max_retries):try:returnhandler(request)exceptExceptionase:ifi==max_retries-1:raiselogger.warning(f"工具调用失败，重试{i+1}/{max_retries}")time.sleep(1)

3. 权限检查：

@wrap_tool_calldefauth_tool_call(request,handler):user_id=request.runtime.context.get("user_id")tool_name=request.tool_call['name']ifnothas_permission(user_id,tool_name):raisePermissionError(f"用户{user_id}无权调用工具{tool_name}")returnhandler(request)

四、实战：模型调用前日志

4.1 使用 @before_model

fromlangchain.agents.middlewareimportbefore_modelfromlangchain.agentsimportAgentStatefromlanggraph.runtimeimportRuntimefromutils.logger_handlerimportlogger@before_modeldeflog_before_model(state:AgentState,# 整个 Agent 的状态记录runtime:Runtime,# 执行过程中的上下文信息):"""在模型执行前输出日志"""logger.info(f"[log_before_model] 即将调用模型，带有{len(state['messages'])}条消息。")# 获取最新消息latest_message=state['messages'][-1]logger.debug(f"[log_before_model]{type(latest_message).__name__}| "f"{latest_message.content.strip()[:100]}")# 返回 None 表示不修改状态returnNone

4.2 修改状态

@before_modeldefinject_context(state,runtime):"""在模型调用前注入上下文信息"""# 添加系统提示state['messages'].insert(0,SystemMessage(content="你是一个专业的客服助手。"))# 返回修改后的状态（或不返回表示不修改）returnstate

4.3 访问运行时上下文

@before_modeldefcheck_report_mode(state,runtime):"""检查是否为报告生成模式"""is_report=runtime.context.get("report",False)ifis_report:logger.info("当前为报告生成场景")# 可以修改 state 或 runtimeelse:logger.info("当前为普通问答场景")returnNone

五、实战：动态提示词切换

5.1 场景说明

需求：同一个 Agent，在不同场景下使用不同的 system_prompt

• 普通问答场景：使用客服提示词
• 报告生成场景：使用报告生成提示词

传统做法：创建两个 Agent

normal_agent=create_agent(model,tools,system_prompt=normal_prompt)report_agent=create_agent(model,tools,system_prompt=report_prompt)

问题：

• 代码重复
• 维护成本高
• 无法运行时切换

中间件方案：一个 Agent，动态切换提示词

5.2 使用 @dynamic_prompt

fromlangchain.agents.middlewareimportdynamic_prompt,ModelRequestfromutils.prompt_loaderimportload_system_prompts,load_report_prompts@dynamic_promptdefreport_prompt_switch(request:ModelRequest):"""动态切换提示词"""# 从运行时上下文获取标记is_report=request.runtime.context.get("report",False)ifis_report:# 报告生成场景，返回报告提示词logger.info("使用报告生成提示词")returnload_report_prompts()# 默认场景，返回系统提示词logger.info("使用系统提示词")returnload_system_prompts()

5.3 完整流程

1. 用户提问："给我生成我的使用报告" ↓ 2. Agent 分析：需要生成报告 ↓ 3. 调用 fill_context_for_report 工具 ↓ 4. monitor_tool 中间件检测到，设置 runtime.context["report"] = True ↓ 5. 调用模型前，dynamic_prompt 中间件被触发 ↓ 6. report_prompt_switch 读取 context["report"] = True ↓ 7. 返回报告生成提示词，注入模型调用 ↓ 8. 模型根据报告提示词生成报告

5.4 标记工具设计

@tool(description="无入参，无返回值，调用后触发中间件自动为报告生成的场景动态注入上下文信息，为后续提示词切换提供上下文信息")deffill_context_for_report():"""标记工具：仅用于触发中间件，不执行实际逻辑"""return"fill_context_for_report 已调用"

设计要点：

• 无实际业务逻辑
• 仅作为"信号"触发中间件
• description 清晰说明用途

六、中间件组合使用

6.1 注册多个中间件

fromlangchain.agentsimportcreate_agent agent=create_agent(model=chat_model,system_prompt=load_system_prompts(),tools=[...],middleware=[monitor_tool,# 工具监控log_before_model,# 模型调用日志report_prompt_switch,# 动态提示词],)

6.2 中间件执行顺序

# 中间件按注册顺序执行middleware=[middleware_a,# 第 1 个执行middleware_b,# 第 2 个执行middleware_c,# 第 3 个执行]

注意：某些钩子的执行顺序可能受 LangChain 内部逻辑影响，建议通过日志验证。

6.3 条件启用中间件

defcreate_agent_with_middleware(scenario="normal"):base_middleware=[monitor_tool,log_before_model]# 根据场景添加特定中间件ifscenario=="report":base_middleware.append(report_prompt_switch)elifscenario=="debug":base_middleware.append(debug_logger)returncreate_agent(model=chat_model,tools=tools,middleware=base_middleware,)

七、高级应用

7.1 安全防护

@wrap_tool_calldefsecurity_check(request,handler):"""安全检查中间件"""# 检查敏感工具sensitive_tools=["delete_user","export_data"]ifrequest.tool_call['name']insensitive_tools:user_id=request.runtime.context.get("user_id")ifnotis_admin(user_id):logger.warning(f"非管理员用户{user_id}尝试调用敏感工具")raisePermissionError("无权执行此操作")# 检查参数中的敏感信息args=request.tool_call.get('args',{})ifcontains_sensitive_data(args):logger.warning("检测到敏感参数")# 可以脱敏或拒绝returnhandler(request)

7.2 性能监控

importtimefromcollectionsimportdefaultdict tool_stats=defaultdict(lambda:{"count":0,"total_time":0})@wrap_tool_calldefperformance_monitor(request,handler):"""性能监控中间件"""start=time.time()result=handler(request)duration=time.time()-start# 统计tool_name=request.tool_call['name']tool_stats[tool_name]["count"]+=1tool_stats[tool_name]["total_time"]+=duration logger.info(f"工具{tool_name}耗时：{duration:.3f}s")returnresult# 定期输出统计defprint_stats():fortool,statsintool_stats.items():avg_time=stats["total_time"]/stats["count"]print(f"{tool}: 调用{stats['count']}次，平均耗时{avg_time:.3f}s")

7.3 缓存机制

fromfunctoolsimportlru_cache cache={}@wrap_tool_calldefcache_tool_call(request,handler):"""工具调用缓存中间件"""tool_name=request.tool_call['name']args=str(request.tool_call.get('args',{}))cache_key=f"{tool_name}:{args}"# 检查缓存ifcache_keyincache:logger.info(f"命中缓存：{cache_key}")returnToolMessage(content=cache[cache_key])# 执行工具result=handler(request)# 写入缓存（只缓存成功结果）cache[cache_key]=result.contentreturnresult

八、调试技巧

8.1 中间件链路日志

@before_agentdeflog_entry(state,runtime):logger.info("=== Agent 执行开始 ===")returnNone@after_agentdeflog_exit(state,runtime):logger.info("=== Agent 执行结束 ===")returnNone@wrap_tool_calldeflog_tool(request,handler):logger.info(f"→ 工具调用：{request.tool_call['name']}")result=handler(request)logger.info(f"← 工具返回：{result.content[:50]}")returnresult

8.2 上下文追踪

@wrap_tool_calldeftrace_context(request,handler):"""追踪运行时上下文变化"""before_context=dict(request.runtime.context)result=handler(request)after_context=dict(request.runtime.context)# 检测上下文变化changes=set(after_context.keys())-set(before_context.keys())ifchanges:logger.info(f"上下文新增键：{changes}")returnresult

九、本章小结

核心要点

1. 中间件类型：

   • 节点式：before_agent, after_agent, before_model, after_model
   • 包装式：wrap_tool_call, wrap_model_call
   • 特殊：dynamic_prompt

2. 典型应用：

   • 工具调用监控
   • 动态提示词切换
   • 安全防护
   • 性能监控
   • 缓存机制

3. 设计原则：

   • 单一职责：每个中间件只做一件事
   • 无副作用：尽量不修改状态
   • 异常处理：捕获并记录异常

下章预告

下一篇我们将讲解RAG 与向量存储，学习：

• Chroma 向量数据库集成
• 文本分片策略
• MD5 去重机制
• 检索器优化

1. Agent 智能体开发实战指南（一）：从 LLM 到 Agent 的认知升级
2. Agent 智能体开发实战指南（二）：工具调用系统深度解析
3. Agent 智能体开发实战指南（三）：ReAct 框架深度解析
4. Agent 智能体开发实战指南（四）：流式输出与状态管理
5. Agent 智能体开发实战指南（五）：中间件系统与动态提示词
6. Agent 智能体开发实战指南（六）：RAG 与向量存储实战
7. Agent 智能体开发实战指南（七）：项目架构设计与工程化实践
8. Agent 智能体开发实战指南（八）：UI 集成与生产部署

本文是《Agent 智能体开发实战指南》系列的第五篇，下一篇将深入讲解 RAG 与向量存储。