昨天我们的代码里，是用 if ai_msg.tool_calls: 来判断并手动执行工具的。 但如果用户的问题是：“帮我查一下北京的天气，然后查一下我日历上今天的日程，根据天气和日程，帮我写一封邮件给老板请假。”

在这个场景里，AI 需要连续调用3次甚至更多次工具（查天气 -> 查日历 -> 发邮件）。如果我们还用手工写 if/else 和 for 循环，代码会变得极其臃肿且难以维护。

解决方案登场：Agent 框架 (以 LangGraph 为例)

今天我们要学的，就是如何使用当今最流行的框架，把昨天的“手动闭环”变成“全自动化驾驶”。只要给 AI 一组工具，它会自动决定调用哪个，自动接收结果，如果不够就继续调用，直到得出最终答案！

前置准备

今天我们会用到目前 LangChain 官方主推的最强 Agent 框架：LangGraph。 请在终端中安装它：

pip install langgraph

Part 1：

实战代码：创建你的第一个全自动 ReAct 智能体

请新建一个文件 demo_auto_agent.py。为了体现全自动的威力，这次我们给 AI 配备两个工具：一个查天气，一个做数学计算。

import os
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
# 🌟 今天的新主角：LangGraph 的预构建 Agent 功能
from langgraph.prebuilt import create_react_agent

# 1. 基础配置
os.environ["OPENAI_API_KEY"] = "sk-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
os.environ["OPENAI_BASE_URL"] = "https://api.openai.com/v1" 

# 推荐 Agent 使用 temperature=0，让它的逻辑推理更严谨、不乱发散
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)

# ==========================================
# 2. 准备工具箱 (给 AI 准备多把武器)
# ==========================================
@tool
def get_weather(city: str) -> str:
    """获取指定城市的天气情况"""
    print(f"\n[后台执行] ☁️ 正在查询 {city} 的天气...")
    if "北京" in city:
        return "晴天，25度"
    return "未知天气"

@tool
def calculate_calories(sport: str, hours: float) -> str:
    """计算某种运动在指定小时内消耗的卡路里"""
    print(f"\n[后台执行] 🧮 正在计算 {sport} 运动 {hours} 小时的卡路里...")
    # 假设每小时消耗 450 卡路里
    calories = hours * 450
    return f"消耗了 {calories} 千卡"

# 把工具打包成列表
tools = [get_weather, calculate_calories]

# ==========================================
# 3. 核心魔法：一键创建全自动 Agent！
# ==========================================
# create_react_agent 会自动帮我们处理“思考->调用工具->等待结果->再思考”的死循环！
agent_executor = create_react_agent(llm, tools)

# ==========================================
# 4. 见证奇迹：提出一个需要“多步思考”的复杂问题
# ==========================================
user_query = "阿泽在北京，他今天想去室外打篮球 2 个小时。请问今天天气合适吗？如果他打完球，能消耗多少卡路里？"

print(f"👤 用户提问: {user_query}\n")
print("-" * 50)
print("🤖 Agent 开始全自动思考与执行（请观察后台日志）：")

# 调用 Agent，输入格式为包含消息列表的字典
response = agent_executor.invoke({"messages": [("user", user_query)]})

print("-" * 50)
print("\n🎉 Agent 最终回答:")
# 打印最后一条消息（即 Agent 的最终总结）
print(response["messages"][-1].content)

为什么叫 ReAct Agent？(今天的核心概念)

当你运行这段代码时，你会发现终端里自动打印出了两次 [后台执行] 的日志，然后才给出最终答案。AI 自动完成了一次复杂的推理！

底层的思维模式叫做 ReAct (Reasoning + Acting，推理与行动)。它的心路历程是这样的：

1. Thought (思考)：用户问了天气和卡路里。我需要先查北京天气。
2. Action (行动)：调用 get_weather，参数 city="北京"。（框架自动执行工具）
3. Observation (观察)：得到结果“晴天，25度”。
4. Thought (再次思考)：天气查到了，适合打球。现在还要算卡路里，我需要调用计算工具。
5. Action (行动)：调用 calculate_calories，参数 sport="打篮球", hours=2。（框架自动执行工具）
6. Observation (观察)：得到结果“消耗了 900 千卡”。
7. Thought (最终思考)：所有信息都收集全了，我可以回答用户了。
8. Final Answer (最终回答)：输出你看到的最终那段自然流畅的文字。

你的任务

1. 安装 langgraph 库。
2. 运行上面这段代码，感受一下不写任何 if-else，Agent 自己连调两个工具的快感。
3. 进阶挑战：你可以试着修改 user_query，比如问它：“我今天在上海打篮球 0.5 个小时消耗多少卡路里？”，看看 Agent 会不会智能地改变它调用的参数。

Part 2：

既然我们已经体验了 create_react_agent 这个“一键生成”的魔法，今天我们就来揭开魔法的黑盒，学习 LangGraph 最核心、最强大的精髓 —— 自定义状态图（StateGraph）。

🧐为什么要揭开黑盒？

前面使用的 create_react_agent 就像是买了一辆“全自动驾驶汽车”，你给目的地，它自己跑。这在简单场景下很好用。 但是，如果你想在流水线中加入自己的逻辑呢？比如：

• 人工审批（Human-in-the-loop）：当 AI 想要调用“转账”或“发邮件”工具时，必须先发一条消息给人类，人类点“同意”，它才能继续执行。
• 多专家协作（Multi-Agent）：比如碰到代码问题交给“程序员大模型”节点，碰到画图问题交给“设计师大模型”节点。

要实现这些高级功能，我们就不能用预设好的 Agent，而是要像画流程图一样，自己把大模型和工具一点一点连接起来。

核心概念：LangGraph 的“三剑客”

我们用“在厨房做一道极其复杂的菜”（或者工厂流水线）来打个比方！

1. State（状态/记忆）—— 传送带上的“菜篮子” 🧺

• 它的角色：它是整个流程中唯一能存储信息的地方。你可以把它想象成在各个工作台之间传递的菜篮子。

• 为什么需要它：因为后面我们要讲的打工人（Node）都是“金鱼记忆”，干完活就忘。所以每次换人干活时，必须把整个菜篮子推给他，他看了篮子里有什么，才知道接下来该怎么干。

• 在代码里：它就是一个字典。在昨天的代码里，这个菜篮子里装着一个名叫 messages 的账本。用户说了什么、大模型想做什么、工具查回来的数据，全都按时间顺序追加（Add）到这个账本里。

2. Node（节点）—— 厨房里的“打工人” 👨‍🍳

• 它的角色：真正在干活的实体。它们只负责一件事：把菜篮子拿过来，做点加工，然后再把菜篮子放回传送带。它们不管推车是从哪来的，也不管下一步要去哪。

• 在代码里，我们安排了两个打工人：

  1. Chatbot Node（主厨·大模型）：大厨拿起菜篮子一看，里面写着“阿泽想去北京打球”。大厨一拍脑袋：“我得查天气！”，于是他在账本上写下“指令：去查北京天气”，然后把篮子放回传送带。

  2. Tool Node（跑腿小弟·工具代码）：他不会自己思考，但他专门负责跑腿。他看到篮子里大厨写的指令，立马跑出厨房去气象局（执行 Python 函数），查到“北京25度晴天”后，写进账本，放回传送带。

3. Edge（边）—— 车间主任的“调度指令” 🚥

• 它的角色：决定菜篮子下一步该推给谁。如果没有 Edge，打工人们干完活，菜篮子就停在原地不动了。

• 在代码里，车间主任有两种调度方式：

  1. 普通边（死规矩）：比如代码里的 graph_builder.add_edge("tools", "chatbot")。意思是：只要“跑腿小弟”拿到了数据，不管三七二十一，必须把篮子推回给“主厨”，让主厨重新过目！

  2. 条件边（智能调度）：也就是那句最神奇的 tools_condition！车间主任会在“主厨”干完活后，凑上去看一眼账本：

     • 情况 A：如果主厨在账本上写了“指令：调用工具”，主任就会一挥手，把篮子推向 Tool Node（跑腿小弟）。

     • 情况 B：如果主厨在账本上写的是“回复阿泽：天气很好，快去打球吧”（没有任何调用工具的请求了），主任就会大手一挥，指向大门 END（结束），把最终的菜端给用户！

实战代码：亲手画出一张 Agent 流程图

这块代码非常经典，它是你通往复杂 AI 架构（如多智能体系统）的必经之路。请新建一个文件 demo_custom_graph.py：

import os
from typing import Annotated
from typing_extensions import TypedDict

from langchain_openai import ChatOpenAI
from langchain_core.tools import tool

# 🌟 引入 LangGraph 核心组件
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition

# 1. 基础配置
os.environ["OPENAI_API_KEY"] = "sk-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
os.environ["OPENAI_BASE_URL"] = "https://api.openai.com/v1"
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)

# ==========================================
# 2. 准备工具箱 (和昨天一样)
# ==========================================
@tool
def get_weather(city: str) -> str:
    """获取指定城市的天气情况"""
    return "晴天，25度" if "北京" in city else "未知天气"

@tool
def calculate_calories(sport: str, hours: float) -> str:
    """计算某种运动在指定小时内消耗的卡路里"""
    return f"消耗了 {hours * 450} 千卡"

tools = [get_weather, calculate_calories]
# 给模型绑定工具
llm_with_tools = llm.bind_tools(tools)

# ==========================================
# 3. 定义 State (传送带上的盒子)
# ==========================================
class State(TypedDict):
    # 我们在这个盒子里放一个名为 messages 的列表。
    # Annotated[list, add_messages] 的意思是：每次有新消息进来，不要覆盖旧消息，而是追加(add)进去。
    messages: Annotated[list, add_messages]

# ==========================================
# 4. 定义 Graph (搭建流水线)
# ==========================================
graph_builder = StateGraph(State)

# 👉 定义第一个节点：AI 思考节点
def chatbot_node(state: State):
    # 这个节点的工作很简单：从传送带(state)拿出历史消息，喂给模型，把模型的回复放回传送带
    response = llm_with_tools.invoke(state["messages"])
    return {"messages": [response]}

# 👉 定义第二个节点：工具执行节点
# ToolNode 是 LangGraph 提供的一个现成节点，它专门负责执行被触发的工具函数
tool_node = ToolNode(tools=tools)

# 把两个节点添加到图纸上
graph_builder.add_node("chatbot", chatbot_node)
graph_builder.add_node("tools", tool_node)

# ==========================================
# 5. 定义 Edge (画连接线，决定工作流向)
# ==========================================
# 规则1：程序的起点永远是 AI 思考节点
graph_builder.add_edge(START, "chatbot")

# 规则2：最核心的“条件边”逻辑
# AI 思考完之后，该去哪？由 tools_condition 这个现成的函数来判断：
# - 如果 AI 返回的消息里包含了工具调用请求（tool_calls），就走向 "tools" 节点
# - 如果 AI 只是普通的聊天回复，就走向 "END" 结束程序
graph_builder.add_conditional_edges("chatbot", tools_condition)

# 规则3：工具节点执行完之后，拿着真实数据，必须回到 AI 思考节点，让 AI 重新总结
graph_builder.add_edge("tools", "chatbot")

# ==========================================
# 6. 编译图纸，生成可执行的 Agent！
# ==========================================
graph = graph_builder.compile()

# ==========================================
# 7. 测试我们的自定义 Graph
# ==========================================
user_query = "阿泽在北京打篮球2小时，天气合适吗？消耗多少卡路里？"
print(f"👤 用户提问: {user_query}\n" + "-" * 50)

# stream() 可以让我们一步步看到这台机器运转的中间过程
for event in graph.stream({"messages": [("user", user_query)]}):
    for node_name, node_state in event.items():
        # 打印当前正在执行的节点名称，以及它产出的最新消息
        latest_message = node_state["messages"][-1]
        print(f"👉 当前执行节点: [{node_name}]")
        
        # 如果是 AI 请求工具调用
        if hasattr(latest_message, "tool_calls") and latest_message.tool_calls:
            print(f"   🤖 AI 请求调用工具: {[t['name'] for t in latest_message.tool_calls]}")
        # 如果是普通文本回复（不论是工具的返回还是 AI 的最终回答）
        elif latest_message.content:
            print(f"   💬 内容: {latest_message.content[:50]}...")
    print("-" * 50)

print("🎉 整个流水线执行完毕！")

💡 运行后你会看到什么？

当你运行这段代码时，你会看到控制台清晰地打印出这台机器的运转轨迹：

1. [chatbot] 节点执行 -> 请求调用工具 get_weather
2. [tools] 节点执行 -> 获取天气结果
3. [chatbot] 节点再执行 -> 觉得还需要查卡路里，请求调用 calculate_calories
4. [tools] 节点再执行 -> 获取卡路里结果
5. [chatbot] 节点最后执行 -> 总结出最终的回答！

这就是为什么它叫 StateGraph (状态图)，你可以极其清晰地看到数据在不同节点之间流转，直到 tools_condition 判断不再需要调用工具，流向 END 为止。