LangGraph 1.x · Plan-and-Execute评估

发表于 更新于 阅读次数： 阅读次数：

1. 背景

E9ovi6

Plan-and-Execute 模式是一种 “先计划、后执行、按计划循环迭代” 的 Agent 设计模式。它的核心思想是：先输出一份结构化的执行计划，将复杂任务拆解为若干明确步骤，然后严格按照步骤逐步执行；当某一步需要外部信息或能力（如搜索、计算等）时，则触发工具调用，完成后继续推进计划，直到满足终止条件。

实际应用场景

以 Cursor 为例：当 Cursor 要编写复杂的代码时，通常会先执行规划，生成一个 to-do list，之后按照这个 to-do list 的顺序依次执行任务。这种”先规划、后执行”的方式能够有效处理复杂任务，避免遗漏关键步骤。

Plan Mode 的三层结构

Plan Mode 通常可以抽象出三层结构：

  1. 计划层：用自然语言清晰列出步骤、编号、依赖、输入输出。
  2. 执行层：逐步消费计划中的步骤，产出结果或中间态。
  3. 工具层：在执行层遇到”需要能力/数据”的节点时，面向具体工具进行调用（如价格查询、表达式计算等）。

优势

这种”先规划、后执行”的范式具备以下优势：

  • 可解释性强
  • 可回溯
  • 易扩展（新增工具即可拓展能力）

在设计 Agent 时已经被广泛采用。

2. 构建 Plan-and-Execute Agent

使用 LangGraph 来构建一个简单的 Plan-and-Execute Agent，节点结构包括：

  • Plan Node（计划节点）
  • Execute Node（执行节点）
  • Tool Node（工具节点）
  • Conditional Edge（条件边）

2.1 基础准备

定义工具

首先开发需要用到的工具（tools）。这里定义了两个函数：一个是查询汇率信息，另一个是计算器，然后利用 LangChain 的 @tool 装饰器，将它们包装成可被 LLM 调用的工具。

代码说明：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
from langchain.tools import tool

@tool
def get_exchange_rate(currency: str) -> str:
    """
    获取货币汇率信息
    
    这个工具用于查询指定货币对人民币的汇率。
    在实际应用中，这里可以对接真实的汇率 API（如 ExchangeRate-API）。
    
    Args:
        currency: 货币代码，例如"USD"（美元）、"EUR"（欧元）
    
    Returns:
        返回格式化的汇率信息字符串
    """
    # 简单的汇率数据映射表（实际应用中应该从汇率API获取）
    if currency == "USD" or currency == "美元":
        return "美元对人民币汇率：1 USD = 7.25 CNY"
    if currency == "EUR" or currency == "欧元":
        return "欧元对人民币汇率：1 EUR = 7.85 CNY"
    return f"没有找到{currency}的汇率信息"

@tool
def calculate(expression: str) -> str:
    """
    计算表达式
    
    这个工具用于执行数学表达式计算。
    注意：实际生产环境中应该使用更安全的表达式解析库（如 numexpr），
    而不是直接使用 eval()，这里仅用于演示。
    
    Args:
        expression: 数学表达式字符串，例如"100 * 7.25"
    
    Returns:
        计算结果字符串
    """
    # 使用 eval 计算表达式（仅用于演示，生产环境需更安全的方案）
    try:
        return str(eval(expression))
    except Exception as e:
        return f"计算错误: {str(e)}"

# 定义工具列表，用于绑定到 LLM
# LLM 可以根据工具描述自动决定何时调用哪个工具
TOOLS = [get_exchange_rate, calculate]

# 创建工具字典，key 为工具名称，value 为工具对象
# 在执行工具调用时，通过工具名称快速查找对应的工具对象
TOOL_DICT = {tool.name: tool for tool in TOOLS}

关键点说明：

  • @tool 装饰器会将普通函数转换为 LangChain 的工具对象，LLM 可以识别并调用
  • 工具函数的文档字符串（docstring）会被 LLM 读取，用于理解工具的用途和参数
  • TOOL_DICT 用于在执行阶段快速查找和调用对应的工具

定义模型

定义大语言模型（LLM），这里使用自建模型（兼容 OpenAI API 格式）。然后将模型与工具进行绑定，使 LLM 能够识别和调用我们定义的工具。

代码说明：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
import os
from dotenv import load_dotenv
from langchain.chat_models import init_chat_model

# 加载 .env 文件中的环境变量
load_dotenv()

# 配置自建模型（必须从 .env 文件中读取）
BASE_URL = os.getenv("OPENAI_BASE_URL")
API_KEY = os.getenv("OPENAI_API_KEY")
MODEL_NAME = os.getenv("OPENAI_MODEL")

# 检查必需的配置项是否存在
if not BASE_URL:
    raise ValueError("OPENAI_BASE_URL 未在 .env 文件中配置")
if not API_KEY:
    raise ValueError("OPENAI_API_KEY 未在 .env 文件中配置")
if not MODEL_NAME:
    raise ValueError("OPENAI_MODEL 未在 .env 文件中配置")

# 创建模型实例
LLM = init_chat_model(
    model=MODEL_NAME,
    model_provider="openai",
    temperature=0.7,
    base_url=BASE_URL,
    api_key=API_KEY,
)

# 将 LLM 与工具列表绑定
# bind_tools() 方法会将工具的描述信息注入到模型的上下文中
# 这样模型在生成回复时，可以识别何时需要调用工具，并生成相应的工具调用请求
LLM_WITH_TOOLS = LLM.bind_tools(TOOLS)

关键点说明：

  • bind_tools() 方法会将工具的函数签名和描述信息传递给 LLM
  • 绑定后，LLM 在生成回复时可能会返回 tool_calls 字段，表示需要调用工具
  • 工具调用信息包含工具名称、参数和调用 ID，用于后续执行

定义状态

定义 Agent 运行过程中需要维护的状态。自定义了 PlanAndExecuteState，它继承 LangGraph 的 MessagesState，可以自动保存消息列表（包括用户消息、AI 回复、工具调用结果等），并且额外扩展了两个属性，分别表示生成的计划和当前执行的步骤。

代码说明：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
from langgraph.graph.message import MessagesState

class PlanAndExecuteState(MessagesState):
    """
    计划和执行状态定义
    
    这个状态类用于在整个 Agent 执行过程中保存和传递信息。
    
    继承自 MessagesState 的属性：
        messages: List[BaseMessage] - 消息列表，包含对话历史、工具调用等
    
    自定义属性：
        plan: str - 生成的执行计划文本
        step: int - 当前执行的步骤编号（从 0 开始）
    """

    plan: str  # 生成的计划，由 plan_node 生成并保存
    step: int  # 当前步骤，每次执行后递增，用于跟踪执行进度

关键点说明：

  • MessagesState 提供了消息管理功能，自动处理消息的追加和更新
  • plan 字段存储计划节点生成的执行计划，供执行节点使用
  • step 字段用于跟踪执行进度，便于调试和日志记录
  • 状态对象在节点之间传递，每个节点可以读取和修改状态

2.2 Plan Node（计划节点）

Plan Node 是整个 Agent 的第一个节点，负责根据用户提出的问题，生成详细的、结构化的执行计划。这个计划将指导后续的执行步骤。

代码说明：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
from langchain.messages import SystemMessage

# 生成计划的 System Prompt
# 这个提示词指导 LLM 如何生成执行计划
SYSTEM_PROMPT_PLAN = """
你是一位智能助手，擅长解决用户提出的各种问题。请为用户提出的问题创建分析方案步骤。

如果有需要，可以调用工具。

你可以调用工具的列表如下：
get_exchange_rate:
    获取指定货币对人民币的汇率信息
    Parameters:
    -----------
    currency : str
        货币代码，例如"USD"（美元）、"EUR"（欧元）
    Returns:
    --------
    str
        指定货币的汇率信息

calculate:
    计算表达式
    Parameters:
    -----------
    expression : str
        表达式内容
    Returns:
    --------
    str
        表达式的计算结果

要求：
1.用中文列出清晰步骤
2.每个步骤标记序号
3.明确说明需要分析和执行的内容
4.只需输出计划内容，不要做任何额外的解释和说明
5.设计的方案步骤要紧紧贴合我的工具所能返回的内容，不要超出工具返回的内容
"""

def plan_node(state: PlanAndExecuteState) -> PlanAndExecuteState:
    """
    生成计划节点
    
    这个节点的职责：
    1. 接收用户的问题（在 state["messages"] 中）
    2. 调用 LLM 生成执行计划
    3. 将计划保存到状态中，供后续节点使用
    
    Args:
        state: 当前状态，包含用户消息
    
    Returns:
        更新后的状态，包含生成的计划
    """
    
    # 构造消息列表
    # SystemMessage 放在最前面，用于设置 LLM 的角色和行为
    # state["messages"] 包含用户的问题
    messages = [SystemMessage(content=SYSTEM_PROMPT_PLAN)] + state["messages"]
    
    # 调用 LLM 生成计划
    # invoke() 方法会发送消息到 LLM，并返回回复
    # .content 获取回复的文本内容（计划文本）
    response = LLM_WITH_TOOLS.invoke(input=messages)
    plan = response.content
    print(f"\n【生成的计划】\n{plan}\n")
    
    # 更新状态，保存生成的计划
    # 这个计划将在 execute_node 中被使用
    state["plan"] = plan
    
    # 返回更新后的状态
    # LangGraph 会将这个状态传递给下一个节点
    return state

工作流程说明：

  1. 节点接收包含用户问题的状态
  2. 构造包含 System Prompt 和用户消息的完整消息列表
  3. 调用 LLM 生成执行计划（纯文本格式，如”1. 查询美元汇率 2. 查询欧元汇率 3. 计算兑换金额”）
  4. 将计划保存到状态中
  5. 返回更新后的状态，流程进入下一个节点

注意：为了方便演示，直接将工具描述信息写到 System Prompt 中了。更理想的方案是接入 MCP（Model Context Protocol）或使用 LangChain 的工具描述自动生成功能，感兴趣的同学可以自己实现。

2.3 Execute Node（执行节点）

Execute Node 是 Agent 的核心执行节点，负责根据生成的计划逐步执行任务。LLM 会根据当前计划、对话历史和工具调用结果，决定下一步操作：可能是直接回复，也可能是调用工具。

代码说明：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
from langchain.messages import SystemMessage

# 执行计划的 System Prompt
# 这个提示词指导 LLM 如何执行计划中的步骤
SYSTEM_PROMPT_EXECUTE = """你是一位思路清晰、有条理的智能助手，你必须严格按照以下计划执行任务：

当前计划：
{plan}

如果你认为计划已经执行到最后一步了，请在内容的末尾加上 Final Answer 字样

示例：
这个问题的最终答案是99.99。Final Answer"""

def execute_node(state: PlanAndExecuteState) -> PlanAndExecuteState:
    """
    执行计划节点
    
    这个节点的职责：
    1. 读取计划节点生成的计划
    2. 根据计划、对话历史和工具调用结果，执行当前步骤
    3. LLM 可能直接回复，也可能生成工具调用请求
    4. 更新执行步骤计数器
    
    Args:
        state: 当前状态，包含计划、消息历史和步骤编号
    
    Returns:
        更新后的状态，包含执行结果
    """
    
    # 获取当前步骤编号
    # 用于跟踪执行进度和日志输出
    step = state["step"]
    
    # 构造消息列表
    # SystemMessage 包含当前计划，指导 LLM 执行
    # state["messages"] 包含完整的对话历史（用户消息、AI回复、工具调用结果等）
    system_message = SystemMessage(content=SYSTEM_PROMPT_EXECUTE.format(plan=state["plan"]))
    messages = [system_message] + state["messages"]
    
    # 调用 LLM 执行计划
    # LLM 会分析当前状态，决定下一步操作
    # 如果 LLM 认为需要调用工具，会在 resp.tool_calls 中返回工具调用信息
    # 如果 LLM 认为可以直接回复，会在 resp.content 中返回文本内容
    resp = LLM_WITH_TOOLS.invoke(input=messages)
    print(f"\n【第 {step + 1} 步】{resp.content}")
    
    # 更新状态
    # 1. 将 LLM 的回复添加到消息历史中
    # 2. 递增步骤计数器
    state["messages"].append(resp)
    state["step"] = step + 1
    
    # 返回更新后的状态
    # 后续的条件边会根据 resp 的内容决定下一步流向
    return state

工作流程说明：

  1. 节点读取状态中的计划（由 plan_node 生成）
  2. 构造包含计划和完整对话历史的消息列表
  3. 调用 LLM，LLM 会：
    • 分析当前应该执行计划的哪一步
    • 如果需要数据，生成工具调用请求（tool_calls 字段）
    • 如果可以直接回答，生成文本回复（content 字段）
  4. 将 LLM 的回复保存到消息历史中
  5. 更新步骤计数器
  6. 返回状态，由条件边决定下一步流向

关键点：

  • LLM 的回复可能包含 tool_calls（需要调用工具）或 content（直接回复）
  • 如果回复末尾包含 “Final Answer”，表示任务完成
  • 步骤计数器用于跟踪执行进度，便于调试

2.4 Tool Node（工具节点）

Tool Node 负责执行工具调用。当 Execute Node 中的 LLM 生成工具调用请求时，流程会进入 Tool Node，执行实际的工具调用，并将结果保存到消息历史中，供后续节点使用。

代码说明：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
from langchain.messages import ToolMessage

def tool_node(state: PlanAndExecuteState) -> PlanAndExecuteState:
    """
    工具调用节点
    
    这个节点的职责：
    1. 从最后一条消息中提取工具调用信息
    2. 根据工具名称查找对应的工具对象
    3. 使用工具参数执行工具调用
    4. 将工具调用结果封装为 ToolMessage，添加到消息历史中
    
    Args:
        state: 当前状态，最后一条消息应包含 tool_calls
    
    Returns:
        更新后的状态，包含工具调用结果
    """
    
    # 获取最后一条消息中的工具调用信息
    # 最后一条消息应该是 execute_node 中 LLM 生成的回复
    # 如果 LLM 决定调用工具，这条消息会包含 tool_calls 字段
    last_message = state["messages"][-1]
    tool_calls = last_message.tool_calls
    
    # 如果没有工具调用，直接返回（理论上不应该发生，因为只有有工具调用才会进入此节点）
    if tool_calls is None:
        return state
    
    # 遍历所有工具调用（LLM 可能同时调用多个工具）
    for tool_call in tool_calls:
        # 提取工具名称和参数
        # tool_call 是一个字典，包含：
        #   - "name": 工具名称，如 "get_exchange_rate"
        #   - "args": 工具参数字典，如 {"currency": "USD"}
        #   - "id": 工具调用 ID，用于关联 ToolMessage
        tool_name = tool_call["name"]
        tool_args = tool_call["args"]
        
        # 从工具字典中查找对应的工具对象
        tool = TOOL_DICT.get(tool_name)
        if tool is None:
            # 如果找不到工具，跳过（实际应用中应该记录错误）
            continue
        
        # 执行工具调用
        # invoke() 方法会使用 tool_args 调用工具函数
        # 例如：get_exchange_rate.invoke({"currency": "USD"}) -> "美元对人民币汇率：1 USD = 7.25 CNY"
        tool_call_result = tool.invoke(input=tool_args)
        print(f"\n调用工具: {tool_name}")
        print(f"参数: {tool_args}")
        print(f"执行结果: {tool_call_result}")
        
        # 将工具调用结果封装为 ToolMessage，添加到消息历史中
        # ToolMessage 需要包含 tool_call_id，用于关联对应的工具调用请求
        # 这样 LLM 在后续回复时，可以知道哪个工具调用产生了哪个结果
        state["messages"].append(
            ToolMessage(
                content=f"工具调用结果: {tool_call_result}",
                tool_call_id=tool_call["id"],  # 关联工具调用请求
            )
        )
    
    # 返回更新后的状态
    # 状态中现在包含了工具调用结果，流程会回到 execute_node 继续执行
    return state

工作流程说明：

  1. 节点从最后一条消息中提取 tool_calls 信息
  2. 遍历每个工具调用请求：
    • 提取工具名称和参数
    • 从 TOOL_DICT 中查找对应的工具对象
    • 使用参数调用工具函数
    • 将结果封装为 ToolMessage 并添加到消息历史
  3. 返回更新后的状态，流程回到 execute_node 继续执行

关键点：

  • ToolMessage 必须包含 tool_call_id，用于关联对应的工具调用请求
  • LLM 在后续回复时，会根据 tool_call_id 匹配工具调用结果
  • 一个 LLM 回复可能包含多个工具调用，需要遍历处理
  • 工具调用结果会被添加到消息历史中，供后续节点使用

2.5 条件边（Conditional Edge）

定义好节点之后，还需要一个 Conditional Edge（条件边）来控制节点之间的流转规则。条件边会根据当前状态决定下一步应该执行哪个节点。

代码说明：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
from langgraph.graph import END

def conditional_edge(state: PlanAndExecuteState) -> str:
    """
    条件边函数
    
    这个函数决定 execute_node 执行后的下一步流向：
    - 如果 LLM 回复中包含 "Final Answer"，返回 END，表示任务完成
    - 否则返回 "tool_node"，表示需要调用工具
    
    Args:
        state: 当前状态，包含执行结果
    
    Returns:
        下一个节点的名称，或 END 表示结束
    """
    
    # 获取最后一条消息（应该是 execute_node 中 LLM 的回复）
    last_message = state["messages"][-1]
    
    # 检查回复中是否包含 "Final Answer" 标识
    # 这是我们在 execute_node 的 System Prompt 中要求的终止条件
    if "Final Answer" in last_message.content:
        # 如果包含，返回 END，表示任务完成，流程结束
        return END
    else:
        # 如果不包含，返回 "tool_node"，表示需要调用工具
        # 注意：这里假设 LLM 的回复一定包含 tool_calls
        # 实际应用中可能需要更复杂的判断逻辑
        return "tool_node"

工作流程说明：

  1. 条件边函数接收当前状态作为参数
  2. 检查最后一条消息（LLM 的回复）中是否包含 “Final Answer”
  3. 如果包含，返回 END，流程结束
  4. 如果不包含，返回 "tool_node"，流程进入工具节点

关键点：

  • 条件边是 LangGraph 中实现分支逻辑的关键机制
  • 返回值必须是已定义的节点名称或 END
  • 这里的判断逻辑比较简单，实际应用中可能需要更复杂的条件判断
  • “Final Answer” 标识是在 System Prompt 中约定的终止条件

2.6 构造 Workflow（工作流）

State、Node 和 Edge 都定义好了之后，就可以构造完整的 Workflow（工作流）了。Workflow 定义了整个 Agent 的执行流程和节点之间的连接关系。

代码说明：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
from langgraph.graph import StateGraph, START, END
from langchain.messages import HumanMessage
from typing_extensions import TypedDict, Annotated

def build_agent():
    """
    构建 Plan-and-Execute Agent
    
    这个函数负责：
    1. 创建状态图（StateGraph）
    2. 添加所有节点
    3. 添加节点之间的边（普通边和条件边）
    4. 编译成可执行的 Agent
    
    Returns:
        编译后的 Agent，可以用于执行任务
    """
    
    # 创建状态图，指定状态类型为 PlanAndExecuteState
    # StateGraph 是 LangGraph 的核心类，用于构建有状态的工作流
    graph = StateGraph(PlanAndExecuteState)
    
    # 添加节点
    # add_node() 方法将节点函数注册到图中
    # 第一个参数是节点名称（字符串），第二个参数是节点函数
    graph.add_node("plan_node", plan_node)      # 计划节点
    graph.add_node("execute_node", execute_node)  # 执行节点
    graph.add_node("tool_node", tool_node)      # 工具节点
    
    # 添加普通边（无条件流转）
    # START 是 LangGraph 的特殊节点，表示工作流的起始点
    graph.add_edge(START, "plan_node")          # 从起始点进入计划节点
    graph.add_edge("plan_node", "execute_node")  # 计划节点执行完后进入执行节点
    
    # 添加条件边（根据条件决定流转方向）
    # add_conditional_edges() 用于添加条件分支
    graph.add_conditional_edges(
        source="execute_node",              # 源节点：执行节点
        path=conditional_edge,              # 条件判断函数
        path_map={                          # 路径映射：条件函数的返回值 -> 目标节点
            "tool_node": "tool_node",      # 如果返回 "tool_node"，进入工具节点
            END: END,                       # 如果返回 END，流程结束
        },
    )
    
    # 添加普通边：工具节点执行完后，回到执行节点继续执行
    # 这样就形成了一个循环：execute_node -> tool_node -> execute_node
    graph.add_edge("tool_node", "execute_node")
    
    # 编译 Agent
    # compile() 方法会将图编译成可执行的状态机
    # 编译后的 Agent 可以接收初始状态并执行整个工作流
    agent = graph.compile()
    
    # 返回编译后的 Agent
    return agent

def run_agent(agent, user_query: str) -> str:  
    """
    运行 Agent
    
    这个函数负责：
    1. 初始化 Agent 状态
    2. 执行 Agent 工作流
    3. 返回最终结果
    
    Args:
        agent: 编译后的 Agent 实例
        user_query: 用户的问题
    
    Returns:
        最终的回答文本
    """
    
    # 初始化状态
    # 创建用户消息，包含用户的问题
    messages = [HumanMessage(content=user_query)]
    
    # 创建初始状态对象
    # plan 初始化为空字符串，将在 plan_node 中生成
    # step 初始化为 0，表示还未开始执行
    # messages 包含用户的问题
    init_state = PlanAndExecuteState(
        plan="",           # 计划初始为空
        step=0,            # 步骤从 0 开始
        messages=messages, # 包含用户消息
    )
    
    # 运行 Agent
    # invoke() 方法会从 START 节点开始执行，直到遇到 END
    # 执行过程中，状态会在节点之间传递和更新
    result = agent.invoke(init_state)
    
    # 返回最终结果
    # 最后一条消息应该是包含 "Final Answer" 的回复
    return result["messages"][-1].content

工作流结构说明：

整个 Agent 的执行流程如下：

1
2
3
4
5
START → plan_node → execute_node → [条件判断]
                                    ↓        ↓
                                 tool_node  END
                                    ↓
                                 execute_node (循环)
  1. START → plan_node：工作流开始，进入计划节点生成计划
  2. plan_node → execute_node：计划生成后，进入执行节点开始执行
  3. execute_node → [条件判断] ：
    • 如果包含 “Final Answer”，流程结束（END）
    • 否则进入 tool_node 执行工具调用
  4. tool_node → execute_node：工具调用完成后，回到执行节点继续执行
  5. 循环执行：步骤 3-4 会循环执行，直到 LLM 输出 “Final Answer”

关键点：

  • StateGraph 是 LangGraph 的核心，用于构建有状态的工作流
  • compile() 方法将图编译成可执行的状态机
  • invoke() 方法执行整个工作流，从 START 到 END
  • 状态在节点之间传递，每个节点可以读取和修改状态
  • 条件边实现了分支逻辑，使工作流能够根据条件选择不同的执行路径

这样，一个完整的 Plan-and-Execute Agent 就构建完成了！

3. 结果演示

测试一下：执行主函数，向 agent 提问 “我想用100美元和50欧元兑换人民币，总共能兑换多少人民币？” 。

代码说明：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
if __name__ == "__main__":
    print("=" * 60)
    print("Plan-and-Execute Agent Demo")
    print("=" * 60)
    
    # 构建 Agent
    agent = build_agent()
    
    # 测试用例：汇率兑换问题
    user_query = "我想用100美元和50欧元兑换人民币，总共能兑换多少人民币？"
    print(f"\n【用户问题】\n{user_query}\n")
    print("-" * 60)
    
    # 运行 Agent
    result = run_agent(agent, user_query)
    
    print("\n" + "=" * 60)
    print("【最终答案】")
    print("=" * 60)
    print(result)
    print("=" * 60)

执行uv run python plan.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
============================================================
Plan-and-Execute Agent Demo
============================================================

【用户问题】
我想用100美元和50欧元兑换人民币，总共能兑换多少人民币？

------------------------------------------------------------

【生成的计划】
1. 调用get_exchange_rate工具，获取美元（USD）对人民币的汇率信息。
2. 调用get_exchange_rate工具，获取欧元（EUR）对人民币的汇率信息。
3. 使用calculate工具，计算100美元兑换成人民币的金额：100 × 美元汇率。
4. 使用calculate工具，计算50欧元兑换成人民币的金额：50 × 欧元汇率。
5. 使用calculate工具，将上述两个兑换金额相加，得出总共能兑换的人民币金额。


【第 1 步】

调用工具: get_exchange_rate
参数: {'currency': 'USD'}
执行结果: 美元对人民币汇率：1 USD = 7.25 CNY

调用工具: get_exchange_rate
参数: {'currency': 'EUR'}
执行结果: 欧元对人民币汇率：1 EUR = 7.85 CNY

【第 2 步】

调用工具: calculate
参数: {'expression': '100 * 7.25'}
执行结果: 725.0

调用工具: calculate
参数: {'expression': '50 * 7.85'}
执行结果: 392.5

【第 3 步】

调用工具: calculate
参数: {'expression': '725.0 + 392.5'}
执行结果: 1117.5

【第 4 步】100美元和50欧元总共能兑换1117.5人民币。Final Answer

============================================================
【最终答案】
============================================================
100美元和50欧元总共能兑换1117.5人民币。Final Answer
============================================================

执行流程说明：

  1. Plan Node：生成计划

    1
    2
    3
    4
    5
    1. 调用get_exchange_rate工具，获取美元（USD）对人民币的汇率信息。
    2. 调用get_exchange_rate工具，获取欧元（EUR）对人民币的汇率信息。
    3. 使用calculate工具，计算100美元兑换成人民币的金额：100 × 美元汇率。
    4. 使用calculate工具，计算50欧元兑换成人民币的金额：50 × 欧元汇率。
    5. 使用calculate工具，将上述两个兑换金额相加，得出总共能兑换的人民币金额。

  2. Execute Node（第1步）：执行计划第1-2步

    • LLM 根据计划，决定同时调用两个 get_exchange_rate 工具
    • 生成两个工具调用请求：查询美元汇率和欧元汇率

  3. Tool Node：执行工具调用

    • 调用 get_exchange_rate({"currency": "USD"})
    • 返回：”美元对人民币汇率：1 USD = 7.25 CNY”
    • 调用 get_exchange_rate({"currency": "EUR"})
    • 返回：”欧元对人民币汇率：1 EUR = 7.85 CNY”

  4. Execute Node（第2步）：执行计划第3-4步

    • LLM 看到两个汇率结果，决定调用 calculate 工具分别计算兑换金额
    • 生成两个工具调用请求：
      • 计算 100 美元：100 * 7.25
      • 计算 50 欧元：50 * 7.85

  5. Tool Node：执行工具调用

    • 调用 calculate({"expression": "100 * 7.25"})
    • 返回：725.0
    • 调用 calculate({"expression": "50 * 7.85"})
    • 返回：392.5

  6. Execute Node（第3步）：执行计划第5步

    • LLM 看到两个兑换金额结果，决定调用 calculate 工具计算总和
    • 生成工具调用请求：725.0 + 392.5

  7. Tool Node：执行工具调用

    • 调用 calculate({"expression": "725.0 + 392.5"})
    • 返回：1117.5

  8. Execute Node（第4步）：生成最终答案

    • LLM 整合所有信息，生成最终答案
    • 输出：”100美元和50欧元总共能兑换1117.5人民币。Final Answer”
    • 条件边检测到 “Final Answer”，流程结束

4. 总结

4.1 Plan-and-Execute 模式的核心流程

Plan-and-Execute 模式的核心流程可以概括为以下五个阶段：

  1. 计划阶段（Plan）：根据用户问题生成结构化的执行计划

    • 将复杂任务拆解为若干明确步骤
    • 每个步骤都有清晰的输入输出说明

  2. 执行阶段（Execute）：按照计划逐步执行

    • LLM 根据当前步骤和上下文决定下一步操作
    • 可能需要调用工具获取外部数据或能力

  3. 工具调用（Tool Calling）：当需要外部能力时，调用相应工具

    • 提取工具调用请求中的参数
    • 执行工具函数并获取结果
    • 将结果保存到消息历史中

  4. 循环迭代（Iteration）：执行完成后判断是否结束

    • 如果未完成，继续执行下一步
    • 工具调用结果会作为上下文传递给下一次执行

  5. 终止条件（Termination）：当检测到终止标识时，任务完成

    • 通过条件边判断是否应该结束
    • 返回最终答案给用户

4.2 模式优势

Plan-and-Execute 模式具有以下优势：

  • 可解释性强：每个步骤都有明确的计划，用户可以清楚地看到 Agent 的执行过程
  • 可回溯：完整的消息历史记录了整个执行过程，便于调试和问题排查
  • 易扩展：新增工具只需定义函数并用 @tool 装饰，无需修改核心逻辑
  • 结构化：计划提供了清晰的任务分解，有助于 LLM 更好地理解和执行任务

4.3 适用场景

这种模式特别适合处理：

  • 复杂的多步骤任务：需要多个步骤才能完成的任务
  • 需要外部数据的任务：需要查询数据库、调用 API 等
  • 需要计算的任务：需要进行数学计算、数据分析等
  • 需要可解释性的任务：用户需要了解 Agent 的执行过程

4.4 改进方向

在实际应用中，可以考虑以下改进：

  • 更智能的计划生成：使用更先进的规划算法，考虑步骤之间的依赖关系
  • 动态计划调整：根据执行结果动态调整计划
  • 错误处理：添加错误处理和重试机制
  • 工具描述自动化：使用 MCP 或自动生成工具描述，而不是手动编写
  • 更复杂的终止条件：不仅依赖 “Final Answer”，还可以考虑步骤完成度、时间限制等

通过 LangGraph 构建 Plan-and-Execute Agent，我们可以轻松实现一个功能强大、可解释、易扩展的智能助手。