LangChain 源码剖析-自定义中间件详解(Custom middleware)

LangChain 源码剖析-自定义中间件详解(Custom middleware)通过实现代理执行流程中特定点运行的钩子来构建自定义中间件。Hook(钩子)中间件提供了两种类型的钩子来拦截代理执行Node-style hooks(节点样式挂钩)在特定执行点按顺序运行。用于日志记录、验证和状态更新。before_agent-在代理启动之前（每次调用一次） before_model-每次模型调用之前 after_model-每个模型响应后 after_agent-代理完成后（每次调用一次）装饰器示例fromlangchain.agents.middlewareimportbefore_model,after_model,AgentStatefromlangchain.messagesimportAIMessagefromlanggraph.runtimeimportRuntimefromtypingimportAny@before_model(can_jump_to=["end"])defcheck_message_limit(state:AgentState,runtime:Runtime)-dict[str,Any]|None:iflen(state["messages"])=50:return{"messages":[AIMessage("Conversation limit reached.")],"jump_to":"end"}returnNone@after_modeldeflog_response(state:AgentState,runtime:Runtime)-dict[str,Any]|None:print(f"Model returned:{state['messages'][-1].content}")returnNone类示例fromlangchain.agents.middlewareimportAgentMiddleware,AgentState,hook_configfromlangchain.messagesimportAIMessagefromlanggraph.runtimeimportRuntimefromtypingimportAnyclassMessageLimitMiddleware(AgentMiddleware):def__init__(self,max_messages:int=50):super().__init__()self.max_messages=max_messages@hook_config(can_jump_to=["end"])defbefore_model(self,state:AgentState,runtime:Runtime)-dict[str,Any]|None:iflen(state["messages"])==self.max_messages:return{"messages":[AIMessage("Conversation limit reached.")],"jump_to":"end"}returnNonedefafter_model(self,state:AgentState,runtime:Runtime)-dict[str,Any]|None:print(f"Model returned:{state['messages'][-1].content}")returnNoneWrap-style hooks(装饰器样式挂钩)@wrap_model_call - 用自定义逻辑包装每个模型调用 @wrap_tool_call - 用自定义逻辑包装每个工具调用Convenience(动态便利性)@dynamic_prompt - 生成动态系统提示示例fromlangchain.agents.middlewareimport(before_model,wrap_model_call,AgentState,ModelRequest,ModelResponse,)fromlangchain.agentsimportcreate_agentfromlanggraph.runtimeimportRuntimefromtypingimportAny,Callable@before_modeldeflog_before_model(state:AgentState,runtime:Runtime)-dict[str,Any]|None:print(f"About to call model with{len(state['messages'])}messages")returnNone@wrap_model_calldefretry_model(request:ModelRequest,handler:Callable[[ModelRequest],ModelResponse],)-ModelResponse:forattemptinrange(3):try:returnhandler(request)exceptExceptionase:ifattempt==2:raiseprint(f"Retry{attempt+1}/3 after error:{e}")agent=create_agent(model=