AI Agent学习 | 结合流程图讲解 can_jump_to 如何在langchain中实现流程的跳转​ 最近看完了某谷的langchain v1.0课程由于这个课程出之前网上基本都是v1.0前的langchain课程有说是1.0版本的课程也是不全的所以那时候就先大概学了一些现在看完这套课程更细化了对langchain的认知。在看的过程中发现can_jump_to这部分算是比较绕的也看到有弹幕说看的迷糊所以自己就写一下来试着讲解这节课的具体内容同时也想巩固一下自己的知识。​ 首先我们先来一起看看这部分的源代码from typing import Any from langchain.agents import create_agent from langchain.agents.middleware import before_model, after_model, AgentState from langchain.messages import AIMessage, SystemMessage from langchain.tools import tool from langgraph.runtime import Runtime tool def get_news() - str: 获取当日新闻 return f美加墨世界杯今日开幕 # 在模型LLM执行前触发。允许跳转到 tools 节点。 before_model(can_jump_to[tools]) def force_tool_first(state: AgentState, runtime: Runtime) - dict[str, Any] | None: 【业务场景强行拦截并触发工具】 如果用户输入包含 direct tool则跳过本次大模型的思考/生成阶段 直接伪造一个大模型的 tool_calls 意图强行把控制权移交给工具执行节点。 text state[messages][-1].content # 检查关键词满足条件则强行干预流程 if isinstance(text, str) and direct tool in text.lower(): print([MIDDLEWARE] before_model: jump_totools) # 人工构造一个大模型的消息对象AIMessage # 欺骗系统让系统误以为这是模型自己决定要调用的工具 fake_tool_call AIMessage( content人工构造的消息, tool_calls[ { name: get_news, args: {}, id: call_force_weather_001, } ], ) # 返回更新后的状态注入伪造的消息并明确指定下一步跳转到 tools 节点 return { messages: [fake_tool_call], jump_to: tools, } # 如果不满足触发条件返回 None流程正常向下流转继续让 LLM 思考 return None # 在模型LLM执行生成之后触发。允许重新跳转回 model 节点。 after_model(can_jump_to[model]) def retry_with_extra_instruction(state: AgentState, runtime: Runtime) - dict[str, Any] | None: 【业务场景反思/重试机制】 如果大模型已经生成了回答但发现用户最初的请求包含 retry model 则动态追加一条系统提示词SystemMessage强行让模型重新生成重试一次。 # 倒序遍历消息历史找到最近的一次用户输入human 消息 user_text for msg in reversed(state[messages]): if getattr(msg, type, ) human: user_text getattr(msg, content, ) break # 检查用户输入是否包含触发重试的关键字 if isinstance(user_text, str) and retry model in user_text.lower(): # 【核心防御】防止无限循环重跳死循环 # 检查消息历史中是否已经注入过这条特殊的系统提示。如果有说明已经重试过了不再重复干预。 already_injected any( isinstance(getattr(msg, content, None), str) and 你必须以【二次回答】开头 in msg.content for msg in state[messages] ) if already_injected: return None # 已注入过直接放行结束重试流程 print([MIDDLEWARE] after_model: jump_tomodel with extra system instruction) # 返回更新后的状态追加强力约束的系统消息并将指针跳回 model 节点重新执行 return { messages: [ SystemMessage(你必须以【二次回答】开头并且只用一句话回答。) ], jump_to: model, } return None # 在模型LLM执行前触发。允许直接跳转到 end 节点强行终止。 before_model(can_jump_to[end]) def overflow_context_processor(state: AgentState, runtime: Runtime) - dict[str, Any] | None: 【业务场景安全卫士/异常拦截】 模拟上下文窗口溢出Token超限或其他严重的系统阻断情况。 一旦触发直接熔断流程拒绝让大模型继续处理直接报错或返回兜底文案。 # 假装溢出,模拟检查最后一条消息是否包含 overflow 标识 if overflow in state[messages][-1].content: print([MIDDLEWARE] before_model: jump_toend when contenxt window overflow) # 构造兜底的结束消息并直接指定跳转到 end 终止 Agent 运行 return { messages: [ AIMessage(上下文窗口溢出终止) ], jump_to: end, } agent create_agent( modelmodel, tools[get_news], # # 将定义的中间件按照顺序挂载到 Agent 中注意执行顺序会严格按照列表声明顺序 middleware[force_tool_first, retry_with_extra_instruction, overflow_context_processor], ) def run_once(user_input: str): result agent.invoke( { messages: [ {role: user, content: user_input} ] } ) for msg in result[messages]: msg.pretty_print() if __name__ __main__: # Case 1: 直接跳 tools # 预期表现 # 1. 触发 force_tool_first打印 [MIDDLEWARE] before_model: jump_totools # 2. 绕过 LLM 的首轮思考直接调用 get_news 工具 # 3. 工具返回结果后LLM 总结工具结果并输出 print( * 30, - Case 1 -, * 30) run_once(请帮我查今日新闻 direct tool) # Case 2: 输出后跳回 model # 预期表现 # 1. 正常进入 LLM 生成第 1 版回答 # 2. 触发 retry_with_extra_instruction打印 [MIDDLEWARE] after_model: jump_tomodel... # 3. 注入系统提示词后LLM 被强行拉回并生成第 2 版回答 # 4. 最终输出应带有“【二次回答】”前缀 print( * 30, - Case 2 -, * 30) run_once(请随便介绍一下 LangChain retry model) # Case 3: # 预期表现 # 1. 触发 overflow_context_processor 中间件 # 2. 直接打印终止信息并退出LLM 根本不会接收到这个请求 print( * 30, - Case 3 -, * 30) run_once(你好 overflow) # Case 4: 正常流程 # 预期表现 # 1. 没有任何中间件被触发不满足任何关键字 # 2. Agent 走正常的 OOTBOut of the box标准工作流User - Model - Call Tool - Model - End print( * 30, - Case 4 -, * 30) run_once(今日新闻摘要)我们先看第一个定义的中间件# 在模型LLM执行前触发。允许跳转到 tools 节点。 before_model(can_jump_to[tools]) def force_tool_first(state: AgentState, runtime: Runtime) - dict[str, Any] | None: 【业务场景强行拦截并触发工具】 如果用户输入包含 direct tool则跳过本次大模型的思考/生成阶段 直接伪造一个大模型的 tool_calls 意图强行把控制权移交给工具执行节点。 text state[messages][-1].content # 检查关键词满足条件则强行干预流程 if isinstance(text, str) and direct tool in text.lower(): print([MIDDLEWARE] before_model: jump_totools) # 人工构造一个大模型的消息对象AIMessage # 欺骗系统让系统误以为这是模型自己决定要调用的工具 fake_tool_call AIMessage( content人工构造的消息, tool_calls[ { name: get_news, args: {}, id: call_force_weather_001, } ], ) # 返回更新后的状态注入伪造的消息并明确指定下一步跳转到 tools 节点 return { messages: [fake_tool_call], jump_to: tools, } # 如果不满足触发条件返回 None流程正常向下流转继续让 LLM 思考 return None首先看一下各个中间件在运行过程中的位置然后回到第一个中间件代码中before_model注解表示调用模型之前触发这个钩子函数can_jump_to[“tools”]的额外参数表示符合条件则跳转到工具调用部分。但是在正常的工具调用流程中一般是由模型根据问题或者需求来选择调用哪些工具然后回返回一个带有tool_calls参数的AIMessage下一步才是工具被调用后返回ToolMessage给模型。但是这里是调用模型前直接跳转的没有涉及到模型的调用要怎么办呢在这个中间件中是通过手动伪造了一个AIMessage来替代大模型完成了工具选择这一部分。if isinstance(text, str) and direct tool in text.lower(): print([MIDDLEWARE] before_model: jump_totools) # 人工构造一个大模型的消息对象AIMessage # 欺骗系统让系统误以为这是模型自己决定要调用的工具 fake_tool_call AIMessage( content人工构造的消息, tool_calls[ { name: get_news, args: {}, id: call_force_weather_001, } ], )所以最后这个中间件被调用时的流程如下接下来到retry_with_extra_instruction这个中间件# 在模型LLM执行生成之后触发。允许重新跳转回 model 节点。 after_model(can_jump_to[model]) def retry_with_extra_instruction(state: AgentState, runtime: Runtime) - dict[str, Any] | None: 【业务场景反思/重试机制】 如果大模型已经生成了回答但发现用户最初的请求包含 retry model 则动态追加一条系统提示词SystemMessage强行让模型重新生成重试一次。 # 倒序遍历消息历史找到最近的一次用户输入human 消息 user_text for msg in reversed(state[messages]): if getattr(msg, type, ) human: user_text getattr(msg, content, ) break # 检查用户输入是否包含触发重试的关键字 if isinstance(user_text, str) and retry model in user_text.lower(): # 【核心防御】防止无限循环重跳死循环 # 检查消息历史中是否已经注入过这条特殊的系统提示。如果有说明已经重试过了不再重复干预。 already_injected any( isinstance(getattr(msg, content, None), str) and 你必须以【二次回答】开头 in msg.content for msg in state[messages] ) if already_injected: return None # 已注入过直接放行结束重试流程 print([MIDDLEWARE] after_model: jump_tomodel with extra system instruction) # 返回更新后的状态追加强力约束的系统消息并将指针跳回 model 节点重新执行 return { messages: [ SystemMessage(你必须以【二次回答】开头并且只用一句话回答。) ], jump_to: model, } return None这个中间件的主要目的就是检测用户输入的提示词中是否有“retry model”这个关键词有的话就会触发can_jump_to这个额外参数再次调用模型生成文本。看一下检测的主要代码# 检查用户输入是否包含触发重试的关键字 if isinstance(user_text, str) and retry model in user_text.lower(): # 【核心防御】防止无限循环重跳死循环 # 检查消息历史中是否已经注入过这条特殊的系统提示。如果有说明已经重试过了不再重复干预。 already_injected any( isinstance(getattr(msg, content, None), str) and 你必须以【二次回答】开头 in msg.content for msg in state[messages] ) if already_injected: return None # 已注入过直接放行结束重试流程 print([MIDDLEWARE] after_model: jump_tomodel with extra system instruction) # 返回更新后的状态追加强力约束的系统消息并将指针跳回 model 节点重新执行 return { messages: [ SystemMessage(你必须以【二次回答】开头并且只用一句话回答。) ], jump_to: model, }可以看到在检测到用户的提示词是否存在关键词retry model后程序会检测消息列表中是否存在有SystemMessage并且其内容和下面的代码注入的内容一样。有的话直接放行避免进入死循环。如果没有则构造一个SystemMessage并注入到模型的系统提示词中并且跳转到模型调用的阶段。注意 : 一般来说模型的system prompt设置好后在调用过程中的消息列表中不会出现但是这里是手动注入的所以在消息列表中可以看到下面的SystemMessage使用pretty_print方法美化输出 System Message 你必须以【二次回答】开头并且只用一句话回答。看到最后一个中间件是模仿上下文窗口溢出后的处理# 在模型LLM执行前触发。允许直接跳转到 end 节点强行终止。 before_model(can_jump_to[end]) def overflow_context_processor(state: AgentState, runtime: Runtime) - dict[str, Any] | None: 【业务场景安全卫士/异常拦截】 模拟上下文窗口溢出Token超限或其他严重的系统阻断情况。 一旦触发直接熔断流程拒绝让大模型继续处理直接报错或返回兜底文案。 # 假装溢出,模拟检查最后一条消息是否包含 overflow 标识 if overflow in state[messages][-1].content: print([MIDDLEWARE] before_model: jump_toend when contenxt window overflow) # 构造兜底的结束消息并直接指定跳转到 end 终止 Agent 运行 return { messages: [ AIMessage(上下文窗口溢出终止) ], jump_to: end, }这里比较简单中间件内写的逻辑就是传给大模型的请求中的消息如果有overflow 标识则直接触发跳转到end节点并构造了一个AIMessage表示模型的响应。对了不要忘记给agent绑定的中间件函数顺序首先先看钩子函数执行顺序统同一钩子下的顺序由绑定顺序决定middleware[force_tool_first, retry_with_extra_instruction, overflow_context_processor],下面是程序的最终输出结合上面的讲解和流程图应该可以较为清晰地解析输出 - Case 1 - [MIDDLEWARE] before_model: jump_totools Human Message 请帮我查今日新闻 direct tool Ai Message 人工构造的消息 Tool Calls: get_news (call_force_weather_001) Call ID: call_force_weather_001 Args: Tool Message Name: get_news 美加墨世界杯今日开幕 Ai Message 今日新闻**美加墨世界杯今日开幕** - Case 2 - [MIDDLEWARE] after_model: jump_tomodel with extra system instruction Human Message 请随便介绍一下 LangChain retry model Ai Message LangChain 里的 **retry model**可以简单理解为**当一次模型调用失败时自动重试的机制**。它的目标不是“让模型更聪明”而是让调用更**稳定、容错**尤其适合遇到网络波动、限流、临时服务异常等情况。 ### 你可以把它理解成这几类场景 - **请求超时**第一次没返回自动再试一次 - **API 限流**被服务端拒绝后等待一会儿再重试 - **临时错误**比如 5xx、连接中断等 - **输出不符合预期**某些链路里会结合校验逻辑失败后重新生成 ### 在 LangChain 里通常怎么用 LangChain 本身常见的是给 **LLM / Runnable / Chain** 包一层重试策略比如 - 设置最大重试次数 - 设置每次重试的等待时间 - 指定哪些异常才重试 - 有时配合指数退避exponential backoff 概念上类似 python chain_with_retry chain.with_retry(stop_after_attempt3) 意思是这条链如果失败最多再试几次。 ### 它的价值 1. **提高稳定性**降低偶发错误带来的失败率 2. **减少人工干预**不用自己手写很多 try/except 3. **更适合生产环境**面对 API 限流、网络抖动更稳 4. **和其他机制互补**比如 fallback、timeout、rate limit 一起用 ### 需要注意的点 - **不是无限重试**否则可能浪费时间和 token - **不要对所有错误都重试**比如参数错误、提示词写错重试也没用 - **重试会增加延迟和成本** - **最好配合日志**方便看是哪一步经常失败 ### 一个直观类比 可以把 retry model 想成“**自动重新拨号**” - 电话没接通不代表你永远打不通 - 但如果号码本身错了打再多次也没用 如果你愿意我还可以继续给你讲 1. **LangChain 的 retry 和 fallback 有什么区别** 2. **Python 里怎么实际写** 3. **和 LangGraph / Runnable 一起怎么用** System Message 你必须以【二次回答】开头并且只用一句话回答。 Ai Message 【二次回答】LangChain 的 retry model 就是给模型调用或链路加一个自动重试机制在超时、限流或临时错误时按设定次数和退避策略重新尝试以提高稳定性。 - Case 3 - [MIDDLEWARE] before_model: jump_toend when contenxt window overflow Human Message 你好 overflow Ai Message 上下文窗口溢出终止 - Case 4 - Human Message 今日新闻摘要 Ai Message Tool Calls: get_news (call_OUPihFl3yxCtn1IWsCv7HHJY) Call ID: call_OUPihFl3yxCtn1IWsCv7HHJY Args: Tool Message Name: get_news 美加墨世界杯今日开幕 Ai Message 今日新闻摘要 - 美加墨世界杯今日开幕 如果你愿意我也可以继续帮你整理成更完整的“体育/国际/财经”分类摘要。上面主要讲解的是使用钩子函数注解的方法来自定义的中间件如果是使用类来实现的话就将原来使用注解的同一钩子函数下的函数体合并成一个方法体作为中间件的一个子方法即可(仅列举部分)class MyMiddleware(AgentMiddleware): hook_config(can_jump_to[tools, end]) def before_model(self, state: AgentState, runtime: Runtime) - dict[str, Any] | None: text state[messages][-1].content # 假装溢出 if overflow in text: print([MIDDLEWARE] before_model: jump_toend when contenxt window overflow) return { messages: [ AIMessage(上下文窗口溢出终止) ], jump_to: end, } if isinstance(text, str) and direct tool in text.lower(): print([MIDDLEWARE] before_model: jump_totools) fake_tool_call AIMessage( content人工构造的消息, tool_calls[ { name: get_news, args: {}, id: call_force_weather_001, } ], ) return { messages: [fake_tool_call], jump_to: tools, } return None ......‘tools’)fake_tool_call AIMessage( content人工构造的消息, tool_calls[ { name: get_news, args: {}, id: call_force_weather_001, } ], ) return { messages: [fake_tool_call], jump_to: tools, } return None ......好了本次讲解就写到这里本人也是初学者第一次发博客如有错误请大家指出( ⁰▿⁰)如果这篇文章有帮助到大家我也很高兴后面也会继续更新学习中遇到的问题解决方法或者知识的总结记录~