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_to='tools'")

        # 人工构造一个大模型的消息对象（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_to='model' 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_to='end' when contenxt window overflow")

        # 构造兜底的结束消息，并直接指定跳转到 "end" 终止 Agent 运行
        return {
            "messages": [
                AIMessage("上下文窗口溢出，终止")
            ],
            "jump_to": "end",
        }


agent = create_agent(
    model=model,
    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_to='tools'"
    # 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_to='model'..."
    # 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 走正常的 OOTB（Out 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_to='tools'")

        # 人工构造一个大模型的消息对象（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_to='tools'")

        # 人工构造一个大模型的消息对象（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_to='model' 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_to='model' 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_to='end' 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_to='tools'
================================ 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_to='model' 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_attempt=3)
```

意思是：这条链如果失败，最多再试几次。

### 它的价值
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_to='end' 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_to='end' 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_to='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
        ......

‘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
    ......

好了本次讲解就写到这里，本人也是初学者，第一次发博客，如有错误请大家指出( ⁰▿⁰)，如果这篇文章有帮助到大家我也很高兴！后面也会继续更新学习中遇到的问题解决方法或者知识的总结记录~