Agent 智能体开发实战指南五中间件系统与动态提示词系列导读这是《Agent 智能体开发实战指南》系列的第五篇将深入讲解 LangChain Agent 的中间件系统包括 Hooks 机制、工具调用监控、动态提示词切换等高级功能让你能够完全掌控 Agent 的执行流程。一、什么是中间件1.1 核心概念中间件Middleware在 Agent 执行过程中拦截和修改行为的组件。类比理解用户请求 → [中间件 1] → [中间件 2] → [Agent] → [中间件 3] → 响应 ↓ ↓ ↓ ↓ 日志记录 权限检查 执行逻辑 结果格式化1.2 为什么需要中间件场景无中间件有中间件日志记录在每个工具中写日志一个中间件统一处理权限检查每个工具单独检查统一拦截验证提示词切换硬编码在 Agent 中动态注入上下文错误处理分散在各处统一捕获处理性能监控手动埋点自动拦截统计1.3 中间件的价值解耦业务逻辑与横切关注点分离复用一个中间件可用于多个 Agent灵活运行时动态添加/移除可观测统一监控和调试二、Hooks 机制详解2.1 六类钩子函数LangChain 提供两类钩子节点式钩子执行点顺序拦截钩子触发时机用途before_agentAgent 执行前预处理、权限检查after_agentAgent 执行后后处理、结果格式化before_model模型调用前修改输入、注入上下文after_model模型调用后修改输出、日志记录包装式钩子针对工具和模型钩子触发时机用途wrap_tool_call工具调用时监控、重试、日志wrap_model_call模型调用时修改模型行为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_callfromtypingimportCallablefromlangchain.agents.middlewareimportwrap_tool_callfromlangchain.tools.tool_nodeimportToolCallRequestfromlangchain_core.messagesimportToolMessagefromutils.logger_handlerimportloggerwrap_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:# 执行工具resulthandler(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)})raisee3.2 关键参数解析ToolCallRequest 结构ToolCallRequest(tool_call{name:get_weather,# 工具名args:{city:深圳},# 参数id:call_abc123# 调用 ID},runtimeRuntime(context{report:False}# 运行时上下文))handler 函数实际执行工具的函数必须调用它否则工具不会执行可以修改参数或结果3.3 应用场景1. 日志记录wrap_tool_calldeflog_tool_call(request,handler):start_timetime.time()resulthandler(request)durationtime.time()-start_time logger.info(f工具{request.tool_call[name]}耗时{duration:.2f}s)returnresult2. 重试机制wrap_tool_calldefretry_tool_call(request,handler):max_retries3foriinrange(max_retries):try:returnhandler(request)exceptExceptionase:ifimax_retries-1:raiselogger.warning(f工具调用失败重试{i1}/{max_retries})time.sleep(1)3. 权限检查wrap_tool_calldefauth_tool_call(request,handler):user_idrequest.runtime.context.get(user_id)tool_namerequest.tool_call[name]ifnothas_permission(user_id,tool_name):raisePermissionError(f用户{user_id}无权调用工具{tool_name})returnhandler(request)四、实战模型调用前日志4.1 使用 before_modelfromlangchain.agents.middlewareimportbefore_modelfromlangchain.agentsimportAgentStatefromlanggraph.runtimeimportRuntimefromutils.logger_handlerimportloggerbefore_modeldeflog_before_model(state:AgentState,# 整个 Agent 的状态记录runtime:Runtime,# 执行过程中的上下文信息):在模型执行前输出日志logger.info(f[log_before_model] 即将调用模型带有{len(state[messages])}条消息。)# 获取最新消息latest_messagestate[messages][-1]logger.debug(f[log_before_model]{type(latest_message).__name__}| f{latest_message.content.strip()[:100]})# 返回 None 表示不修改状态returnNone4.2 修改状态before_modeldefinject_context(state,runtime):在模型调用前注入上下文信息# 添加系统提示state[messages].insert(0,SystemMessage(content你是一个专业的客服助手。))# 返回修改后的状态或不返回表示不修改returnstate4.3 访问运行时上下文before_modeldefcheck_report_mode(state,runtime):检查是否为报告生成模式is_reportruntime.context.get(report,False)ifis_report:logger.info(当前为报告生成场景)# 可以修改 state 或 runtimeelse:logger.info(当前为普通问答场景)returnNone五、实战动态提示词切换5.1 场景说明需求同一个 Agent在不同场景下使用不同的 system_prompt普通问答场景使用客服提示词报告生成场景使用报告生成提示词传统做法创建两个 Agentnormal_agentcreate_agent(model,tools,system_promptnormal_prompt)report_agentcreate_agent(model,tools,system_promptreport_prompt)问题代码重复维护成本高无法运行时切换中间件方案一个 Agent动态切换提示词5.2 使用 dynamic_promptfromlangchain.agents.middlewareimportdynamic_prompt,ModelRequestfromutils.prompt_loaderimportload_system_prompts,load_report_promptsdynamic_promptdefreport_prompt_switch(request:ModelRequest):动态切换提示词# 从运行时上下文获取标记is_reportrequest.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():标记工具仅用于触发中间件不执行实际逻辑returnfill_context_for_report 已调用设计要点无实际业务逻辑仅作为信号触发中间件description 清晰说明用途六、中间件组合使用6.1 注册多个中间件fromlangchain.agentsimportcreate_agent agentcreate_agent(modelchat_model,system_promptload_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(scenarionormal):base_middleware[monitor_tool,log_before_model]# 根据场景添加特定中间件ifscenarioreport:base_middleware.append(report_prompt_switch)elifscenariodebug:base_middleware.append(debug_logger)returncreate_agent(modelchat_model,toolstools,middlewarebase_middleware,)七、高级应用7.1 安全防护wrap_tool_calldefsecurity_check(request,handler):安全检查中间件# 检查敏感工具sensitive_tools[delete_user,export_data]ifrequest.tool_call[name]insensitive_tools:user_idrequest.runtime.context.get(user_id)ifnotis_admin(user_id):logger.warning(f非管理员用户{user_id}尝试调用敏感工具)raisePermissionError(无权执行此操作)# 检查参数中的敏感信息argsrequest.tool_call.get(args,{})ifcontains_sensitive_data(args):logger.warning(检测到敏感参数)# 可以脱敏或拒绝returnhandler(request)7.2 性能监控importtimefromcollectionsimportdefaultdict tool_statsdefaultdict(lambda:{count:0,total_time:0})wrap_tool_calldefperformance_monitor(request,handler):性能监控中间件starttime.time()resulthandler(request)durationtime.time()-start# 统计tool_namerequest.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_timestats[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_namerequest.tool_call[name]argsstr(request.tool_call.get(args,{}))cache_keyf{tool_name}:{args}# 检查缓存ifcache_keyincache:logger.info(f命中缓存{cache_key})returnToolMessage(contentcache[cache_key])# 执行工具resulthandler(request)# 写入缓存只缓存成功结果cache[cache_key]result.contentreturnresult八、调试技巧8.1 中间件链路日志before_agentdeflog_entry(state,runtime):logger.info( Agent 执行开始 )returnNoneafter_agentdeflog_exit(state,runtime):logger.info( Agent 执行结束 )returnNonewrap_tool_calldeflog_tool(request,handler):logger.info(f→ 工具调用{request.tool_call[name]})resulthandler(request)logger.info(f← 工具返回{result.content[:50]})returnresult8.2 上下文追踪wrap_tool_calldeftrace_context(request,handler):追踪运行时上下文变化before_contextdict(request.runtime.context)resulthandler(request)after_contextdict(request.runtime.context)# 检测上下文变化changesset(after_context.keys())-set(before_context.keys())ifchanges:logger.info(f上下文新增键{changes})returnresult九、本章小结核心要点中间件类型节点式before_agent, after_agent, before_model, after_model包装式wrap_tool_call, wrap_model_call特殊dynamic_prompt典型应用工具调用监控动态提示词切换安全防护性能监控缓存机制设计原则单一职责每个中间件只做一件事无副作用尽量不修改状态异常处理捕获并记录异常下章预告下一篇我们将讲解RAG 与向量存储学习Chroma 向量数据库集成文本分片策略MD5 去重机制检索器优化Agent 智能体开发实战指南一从 LLM 到 Agent 的认知升级Agent 智能体开发实战指南二工具调用系统深度解析Agent 智能体开发实战指南三ReAct 框架深度解析Agent 智能体开发实战指南四流式输出与状态管理Agent 智能体开发实战指南五中间件系统与动态提示词Agent 智能体开发实战指南六RAG 与向量存储实战Agent 智能体开发实战指南七项目架构设计与工程化实践Agent 智能体开发实战指南八UI 集成与生产部署本文是《Agent 智能体开发实战指南》系列的第五篇下一篇将深入讲解 RAG 与向量存储。
05-Agent 智能体开发实战指南（五）：中间件系统与动态提示词
Agent 智能体开发实战指南五中间件系统与动态提示词系列导读这是《Agent 智能体开发实战指南》系列的第五篇将深入讲解 LangChain Agent 的中间件系统包括 Hooks 机制、工具调用监控、动态提示词切换等高级功能让你能够完全掌控 Agent 的执行流程。一、什么是中间件1.1 核心概念中间件Middleware在 Agent 执行过程中拦截和修改行为的组件。类比理解用户请求 → [中间件 1] → [中间件 2] → [Agent] → [中间件 3] → 响应 ↓ ↓ ↓ ↓ 日志记录 权限检查 执行逻辑 结果格式化1.2 为什么需要中间件场景无中间件有中间件日志记录在每个工具中写日志一个中间件统一处理权限检查每个工具单独检查统一拦截验证提示词切换硬编码在 Agent 中动态注入上下文错误处理分散在各处统一捕获处理性能监控手动埋点自动拦截统计1.3 中间件的价值解耦业务逻辑与横切关注点分离复用一个中间件可用于多个 Agent灵活运行时动态添加/移除可观测统一监控和调试二、Hooks 机制详解2.1 六类钩子函数LangChain 提供两类钩子节点式钩子执行点顺序拦截钩子触发时机用途before_agentAgent 执行前预处理、权限检查after_agentAgent 执行后后处理、结果格式化before_model模型调用前修改输入、注入上下文after_model模型调用后修改输出、日志记录包装式钩子针对工具和模型钩子触发时机用途wrap_tool_call工具调用时监控、重试、日志wrap_model_call模型调用时修改模型行为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_callfromtypingimportCallablefromlangchain.agents.middlewareimportwrap_tool_callfromlangchain.tools.tool_nodeimportToolCallRequestfromlangchain_core.messagesimportToolMessagefromutils.logger_handlerimportloggerwrap_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:# 执行工具resulthandler(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)})raisee3.2 关键参数解析ToolCallRequest 结构ToolCallRequest(tool_call{name:get_weather,# 工具名args:{city:深圳},# 参数id:call_abc123# 调用 ID},runtimeRuntime(context{report:False}# 运行时上下文))handler 函数实际执行工具的函数必须调用它否则工具不会执行可以修改参数或结果3.3 应用场景1. 日志记录wrap_tool_calldeflog_tool_call(request,handler):start_timetime.time()resulthandler(request)durationtime.time()-start_time logger.info(f工具{request.tool_call[name]}耗时{duration:.2f}s)returnresult2. 重试机制wrap_tool_calldefretry_tool_call(request,handler):max_retries3foriinrange(max_retries):try:returnhandler(request)exceptExceptionase:ifimax_retries-1:raiselogger.warning(f工具调用失败重试{i1}/{max_retries})time.sleep(1)3. 权限检查wrap_tool_calldefauth_tool_call(request,handler):user_idrequest.runtime.context.get(user_id)tool_namerequest.tool_call[name]ifnothas_permission(user_id,tool_name):raisePermissionError(f用户{user_id}无权调用工具{tool_name})returnhandler(request)四、实战模型调用前日志4.1 使用 before_modelfromlangchain.agents.middlewareimportbefore_modelfromlangchain.agentsimportAgentStatefromlanggraph.runtimeimportRuntimefromutils.logger_handlerimportloggerbefore_modeldeflog_before_model(state:AgentState,# 整个 Agent 的状态记录runtime:Runtime,# 执行过程中的上下文信息):在模型执行前输出日志logger.info(f[log_before_model] 即将调用模型带有{len(state[messages])}条消息。)# 获取最新消息latest_messagestate[messages][-1]logger.debug(f[log_before_model]{type(latest_message).__name__}| f{latest_message.content.strip()[:100]})# 返回 None 表示不修改状态returnNone4.2 修改状态before_modeldefinject_context(state,runtime):在模型调用前注入上下文信息# 添加系统提示state[messages].insert(0,SystemMessage(content你是一个专业的客服助手。))# 返回修改后的状态或不返回表示不修改returnstate4.3 访问运行时上下文before_modeldefcheck_report_mode(state,runtime):检查是否为报告生成模式is_reportruntime.context.get(report,False)ifis_report:logger.info(当前为报告生成场景)# 可以修改 state 或 runtimeelse:logger.info(当前为普通问答场景)returnNone五、实战动态提示词切换5.1 场景说明需求同一个 Agent在不同场景下使用不同的 system_prompt普通问答场景使用客服提示词报告生成场景使用报告生成提示词传统做法创建两个 Agentnormal_agentcreate_agent(model,tools,system_promptnormal_prompt)report_agentcreate_agent(model,tools,system_promptreport_prompt)问题代码重复维护成本高无法运行时切换中间件方案一个 Agent动态切换提示词5.2 使用 dynamic_promptfromlangchain.agents.middlewareimportdynamic_prompt,ModelRequestfromutils.prompt_loaderimportload_system_prompts,load_report_promptsdynamic_promptdefreport_prompt_switch(request:ModelRequest):动态切换提示词# 从运行时上下文获取标记is_reportrequest.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():标记工具仅用于触发中间件不执行实际逻辑returnfill_context_for_report 已调用设计要点无实际业务逻辑仅作为信号触发中间件description 清晰说明用途六、中间件组合使用6.1 注册多个中间件fromlangchain.agentsimportcreate_agent agentcreate_agent(modelchat_model,system_promptload_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(scenarionormal):base_middleware[monitor_tool,log_before_model]# 根据场景添加特定中间件ifscenarioreport:base_middleware.append(report_prompt_switch)elifscenariodebug:base_middleware.append(debug_logger)returncreate_agent(modelchat_model,toolstools,middlewarebase_middleware,)七、高级应用7.1 安全防护wrap_tool_calldefsecurity_check(request,handler):安全检查中间件# 检查敏感工具sensitive_tools[delete_user,export_data]ifrequest.tool_call[name]insensitive_tools:user_idrequest.runtime.context.get(user_id)ifnotis_admin(user_id):logger.warning(f非管理员用户{user_id}尝试调用敏感工具)raisePermissionError(无权执行此操作)# 检查参数中的敏感信息argsrequest.tool_call.get(args,{})ifcontains_sensitive_data(args):logger.warning(检测到敏感参数)# 可以脱敏或拒绝returnhandler(request)7.2 性能监控importtimefromcollectionsimportdefaultdict tool_statsdefaultdict(lambda:{count:0,total_time:0})wrap_tool_calldefperformance_monitor(request,handler):性能监控中间件starttime.time()resulthandler(request)durationtime.time()-start# 统计tool_namerequest.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_timestats[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_namerequest.tool_call[name]argsstr(request.tool_call.get(args,{}))cache_keyf{tool_name}:{args}# 检查缓存ifcache_keyincache:logger.info(f命中缓存{cache_key})returnToolMessage(contentcache[cache_key])# 执行工具resulthandler(request)# 写入缓存只缓存成功结果cache[cache_key]result.contentreturnresult八、调试技巧8.1 中间件链路日志before_agentdeflog_entry(state,runtime):logger.info( Agent 执行开始 )returnNoneafter_agentdeflog_exit(state,runtime):logger.info( Agent 执行结束 )returnNonewrap_tool_calldeflog_tool(request,handler):logger.info(f→ 工具调用{request.tool_call[name]})resulthandler(request)logger.info(f← 工具返回{result.content[:50]})returnresult8.2 上下文追踪wrap_tool_calldeftrace_context(request,handler):追踪运行时上下文变化before_contextdict(request.runtime.context)resulthandler(request)after_contextdict(request.runtime.context)# 检测上下文变化changesset(after_context.keys())-set(before_context.keys())ifchanges:logger.info(f上下文新增键{changes})returnresult九、本章小结核心要点中间件类型节点式before_agent, after_agent, before_model, after_model包装式wrap_tool_call, wrap_model_call特殊dynamic_prompt典型应用工具调用监控动态提示词切换安全防护性能监控缓存机制设计原则单一职责每个中间件只做一件事无副作用尽量不修改状态异常处理捕获并记录异常下章预告下一篇我们将讲解RAG 与向量存储学习Chroma 向量数据库集成文本分片策略MD5 去重机制检索器优化Agent 智能体开发实战指南一从 LLM 到 Agent 的认知升级Agent 智能体开发实战指南二工具调用系统深度解析Agent 智能体开发实战指南三ReAct 框架深度解析Agent 智能体开发实战指南四流式输出与状态管理Agent 智能体开发实战指南五中间件系统与动态提示词Agent 智能体开发实战指南六RAG 与向量存储实战Agent 智能体开发实战指南七项目架构设计与工程化实践Agent 智能体开发实战指南八UI 集成与生产部署本文是《Agent 智能体开发实战指南》系列的第五篇下一篇将深入讲解 RAG 与向量存储。
