1. 从LangChain到LangGraph：为什么我们需要“图”来构建Agent？

如果你之前玩过LangChain，用它来调用大模型、连接工具，那你肯定体验过那种“链式”编程的爽快感。一条链，一个接一个地调用，逻辑清晰。但不知道你有没有遇到过这种情况：当任务稍微复杂一点，比如需要根据上一步的结果来决定下一步是调用工具A还是工具B，或者需要循环执行某个步骤直到满足条件，单纯的“链”就开始有点力不从心了。代码里会开始出现一堆 if...else 判断，逻辑变得像一团乱麻，可读性和可维护性直线下降。

这时候，LangGraph 就该登场了。你可以把它理解为 LangChain 的“超级赛亚人”形态，是专门为处理有状态、多分支、可循环的复杂工作流而生的。它把整个Agent的执行过程抽象成一张有向图，图中的节点（Node）代表一个执行单元（比如调用大模型做决策、执行某个工具），边（Edge）代表执行流的方向。这种图形化的思维方式，和我们大脑处理复杂问题的逻辑非常像，所以一旦理解，你会觉得无比清晰。

我刚开始接触时也觉得有点抽象，后来想了个生活化的比喻：如果把构建一个智能助手比作设计一个自动化厨房。

• LangChain 就像一套预设好的线性菜谱：先洗菜（节点1），再切菜（节点2），然后开火炒（节点3）。步骤固定，顺序固定。
• LangGraph 则像是一个智能厨房调度系统。它有一个中央大脑（Agent节点），这个大脑会随时根据“当前状态”（比如食材准备好了没、锅热了没）来决定下一步是让“切菜机器人”（工具节点A）工作，还是让“测温仪”（工具节点B）工作。并且，炒完菜后，大脑可能还会尝一下味道（回到决策节点），如果咸了，就命令“加水工具”（工具节点C）执行一下。这个过程可以循环，直到味道满意为止。

所以，LangGraph的核心价值，在于它完美地管理了“状态”和“流程控制”。它帮你把“现在进行到哪一步了”、“之前都干过什么”、“接下来该干嘛”这些琐碎但又至关重要的状态信息封装起来，让你能更专注于定义每个环节的具体逻辑。接下来，我们就亲手搭建一个这样的“智能厨房”，构建一个能自主调用工具的Agent。

2. 实战准备：定义你的工具与智能大脑

在开始画图（构建Graph）之前，我们得先把图里的“零件”准备好。主要就是两样东西：Agent能用的工具（Tools），和做决策的大脑（LLM）。

2.1 打造趁手的工具

工具其实就是一个个函数，它们能完成一些LLM不擅长的事情，比如计算、查数据库、调用外部API。在LangChain的体系里，用 @tool 装饰器来定义一个工具是最方便的方式。

我们这里定义三个工具，覆盖几种常见类型：

1. 一个纯逻辑处理的工具：比如分析文本情感和语言。
2. 一个信息摘要工具：对文本进行概述。
3. 一个调用外部API的工具：获取实时天气。

这里我踩过一个小坑：工具函数的描述（docstring）和参数描述一定要尽可能清晰！因为大模型就是靠这些描述来理解“该在什么时候、用什么参数来调用这个工具”。描述模糊，Agent就容易犯傻。

import datetime
import requests
from langchain.tools import tool
from langchain.pydantic_v1 import BaseModel, Field

# 工具1：情感与语言分析工具
class TaggingSchema(BaseModel):
    """分析句子的情感极性，并输出句子对应的语言"""
    sentiment: str = Field(description="文本的情感，应为 `pos`（积极）、`neg`（消极）或 `neutral`（中性）")
    language: str = Field(description="文本的语言（应使用 ISO 639-1 代码，如 zh, en）")

@tool("tagging", args_schema=TaggingSchema)
def tagging_tool(sentiment: str, language: str) -> str:
    """
    根据给定的情感和语言参数，生成描述语句。
    这是一个示例工具，实际应用中可以接入真实的情感分析模型。
    """
    return f"分析完成：情感倾向为【{sentiment}】，使用语言为【{language}】。"

# 工具2：文本概述工具
class OverviewSchema(BaseModel):
    """对一段文本进行概述"""
    summary: str = Field(description="提供内容的简洁摘要。")
    language: str = Field(description="提供内容所使用的语言。")
    keywords: str = Field(description="提供与内容相关的关键词。")

@tool("overview", args_schema=OverviewSchema)
def overview_tool(summary: str, language: str, keywords: str) -> str:
    """接收摘要、语言和关键词，格式化输出。"""
    return f"概述总结：\n摘要：{summary}\n语言：{language}\n关键词：{keywords}"

# 工具3：实时天气查询工具（调用真实API）
@tool
def get_current_temperature(latitude: float, longitude: float) -> str:
    """根据经纬度坐标获取当前的温度。"""
    BASE_URL = "https://api.open-meteo.com/v1/forecast"
    params = {
        'latitude': latitude,
        'longitude': longitude,
        'hourly': 'temperature_2m',
        'forecast_days': 1,
    }
    response = requests.get(BASE_URL, params=params)
    if response.status_code == 200:
        results = response.json()
    else:
        raise Exception(f"天气API请求失败，状态码：{response.status_code}")

    # 找到最接近当前UTC时间的数据点
    current_utc_time = datetime.datetime.utcnow()
    time_list = [datetime.datetime.fromisoformat(t.replace('Z', '+00:00')) for t in results['hourly']['time']]
    temp_list = results['hourly']['temperature_2m']
    closest_idx = min(range(len(time_list)), key=lambda i: abs(time_list[i] - current_utc_time))
    current_temp = temp_list[closest_idx]
    return f'当前位置（纬度{latitude}, 经度{longitude}）的当前温度是 {current_temp}°C'

# 将工具放入列表，供后续使用
tools = [tagging_tool, overview_tool, get_current_temperature]

2.2 接入大模型作为决策核心

有了工具，还需要一个“指挥官”。这里我们使用智谱AI的GLM模型，因为它对OpenAI的 tools 调用格式兼容得很好，国内访问也稳定。当然，你也可以轻松替换成OpenAI或其他兼容API。

from langchain.chat_models import ChatOpenAI
from langchain.agents import create_openai_tools_agent
from langchain import hub
import os

# 假设你的API Key已设置在环境变量中
os.environ["ZHIPUAI_API_KEY"] = "your_api_key_here"

# 初始化模型。注意：这里使用 `ChatOpenAI` 基类，但通过 `base_url` 和 `model` 参数指向智谱
llm = ChatOpenAI(
    model="glm-3-turbo", # 或 "glm-4"
    base_url="https://open.bigmodel.cn/api/paas/v4/",
    api_key=os.environ["ZHIPUAI_API_KEY"],
    temperature=0.1 # 对于工具调用，温度可以设低一点，让决策更稳定
)

# 使用一个社区共享的、针对工具调用优化过的Prompt模板
prompt = hub.pull("hwchase17/openai-tools-agent")

# 创建Agent的核心可运行对象（Runnable）
# 这个 `agent_runnable` 就是我们的“大脑”，它内部集成了LLM、Prompt和对工具定义的感知
agent_runnable = create_openai_tools_agent(llm, tools, prompt)

到这里，零件备齐了。但光有零件不行，我们还需要一个“流水线”和“调度手册”来规定它们如何协作。这就是LangGraph要管理的状态和图结构。

3. 核心设计：定义Agent的状态与图节点

LangGraph的 StatefulGraph 是其灵魂所在。它要求我们显式地定义一个状态（State）字典，这个字典会在整个图的各个节点间传递和更新。这就像给我们的智能厨房配了一个共享白板，每个工作环节（节点）都可以在上面读取信息、写下自己的工作结果。

3.1 设计状态蓝图

我们需要仔细想想，在整个Agent运行过程中，有哪些信息是需要持久保存、并在节点间共享的。对于一个典型的工具调用Agent，我通常会定义以下几个核心字段：

from typing import TypedDict, Union, Annotated
from langchain_core.agents import AgentAction, AgentFinish
from langchain_core.messages import BaseMessage
import operator

class AgentState(TypedDict):
    """
    Agent运行过程中的全局状态。
    这是一个类型字典，定义了状态的“形状”。
    """
    # 用户当前轮次的输入问题
    input: str
    # 历史对话消息列表，用于让Agent有上下文记忆
    chat_history: list[BaseMessage]
    # Agent节点的输出结果。可能是下一步要执行的Action，也可能是表示结束的Finish。
    # 初始状态为 None
    agent_outcome: Union[AgentAction, AgentFinish, None]
    # 已执行的（动作，观察结果）对列表。
    # `Annotated` 和 `operator.add` 是关键！它告诉LangGraph，当多个节点更新这个字段时，
    # 应该用“追加”的方式，而不是覆盖。这完美记录了执行轨迹。
    intermediate_steps: Annotated[list[tuple[AgentAction, str]], operator.add]

这个 AgentState 类就像我们设计的共享白板的表格模板。input 和 chat_history 是输入信息。agent_outcome 是“大脑”刚刚做出的决策。intermediate_steps 是所有已执行动作和结果的流水账，它用 operator.add 标注，意味着每个节点都可以往这个列表里添加新记录，而不会抹掉之前的。这个设计非常巧妙，省去了我们自己维护列表合并的麻烦。

3.2 创建两个核心工作节点

图由节点构成。我们这个简单Agent主要有两类节点：

1. Agent节点：负责“思考”。它查看当前状态（用户问题、历史、已执行步骤），然后决定下一步是调用某个工具，还是认为任务已完成可以回复用户。
2. 工具执行节点：负责“动手”。它忠实地执行Agent节点发出的工具调用指令，并把结果记录下来。

from langgraph.prebuilt.tool_executor import ToolExecutor

# ToolExecutor 是一个LangGraph提供的工具执行器，非常方便
tool_executor = ToolExecutor(tools)

# 定义 Agent 节点函数
def run_agent_node(state: AgentState) -> dict:
    """
    Agent思考节点。
    接收当前状态，调用agent_runnable进行决策，返回更新了`agent_outcome`的状态片段。
    """
    # agent_runnable.invoke 需要的是一个包含所有状态键的字典
    agent_outcome = agent_runnable.invoke(state)
    # 返回的字典中的键值对，会被LangGraph自动合并（merge）到全局状态中
    return {"agent_outcome": agent_outcome}

# 定义 工具执行 节点函数
def execute_tools_node(state: AgentState) -> dict:
    """
    工具执行节点。
    从状态中取出最新的`agent_outcome`（应该是一个AgentAction），执行对应的工具。
    """
    # 取出大脑的决策结果
    agent_action = state["agent_outcome"]
    # 这里假设agent_action是一个包含多个动作的列表，我们取最后一个（对于OpenAI格式，通常如此）
    # 实际调试时，建议打印一下 agent_action 的结构，以便理解
    print(f"[工具节点] 正在执行动作: {agent_action}")
    # 调用工具执行器
    observation = tool_executor.invoke(agent_action)
    # 将本次执行的（动作，观察结果）对，追加到 intermediate_steps 中
    return {"intermediate_steps": [(agent_action, observation)]}

注意：run_agent_node 返回 {"agent_outcome": ...}，而 execute_tools_node 返回 {"intermediate_steps": ...}。LangGraph会自动将这些返回的字典与当前状态进行合并更新。这就是状态管理的魔力。

4. 组装与调度：构建图的工作流

零件（工具、模型）有了，工作台（状态）有了，工人（节点函数）也有了。现在我们需要画出生产线流程图，规定好工人的工作顺序和交接规则。这就是构建 StateGraph。

4.1 创建图并添加节点

from langgraph.graph import StateGraph, END

# 1. 创建一个以 AgentState 为状态类型的图
workflow = StateGraph(AgentState)

# 2. 添加两个节点，并给它们起名
workflow.add_node("agent", run_agent_node)   # “思考者”节点
workflow.add_node("action", execute_tools_node) # “执行者”节点

4.2 设定流程走向：边与条件边

这是最体现“图”的智能的地方。我们需要定义：

• 入口点：工作从哪个节点开始。
• 普通边：无条件地从一个节点到另一个节点。
• 条件边：根据当前状态的值，动态决定下一个节点是谁。

# 3. 设定入口点：工作先从“agent”节点开始（先思考）
workflow.set_entry_point("agent")

# 4. 定义一个路由判断函数
def should_continue(state: AgentState) -> str:
    """
    根据Agent的决策结果，决定图该往哪走。
    返回一个字符串，对应后续边的‘键’。
    """
    outcome = state["agent_outcome"]
    # 如果大脑说任务完成了（返回AgentFinish），我们就结束流程
    if isinstance(outcome, AgentFinish):
        return "end"
    # 否则，大脑说需要执行一个动作（返回AgentAction），我们就继续去执行工具
    else:
        return "continue"

# 5. 为“agent”节点添加【条件边】
workflow.add_conditional_edges(
    "agent", # 源节点
    should_continue, # 路由判断函数
    {
        "continue": "action", # 如果返回"continue"，下一步去"action"节点
        "end": END           # 如果返回"end"，则图运行结束
    }
)

# 6. 添加一条从“action”节点回到“agent”节点的【普通边】
# 这意味着每次执行完工具后，必须再次回到“大脑”节点进行下一轮思考，形成循环。
workflow.add_edge("action", "agent")

我来解释一下这个流程，它形成了一个经典的 “思考-行动-再思考” 循环：

1. 用户输入问题，图从 agent 节点开始。
2. agent 节点思考后，产生一个结果 (agent_outcome)。
3. should_continue 函数检查这个结果：如果是 AgentFinish，直接走到 END，流程结束，返回最终答案；如果是 AgentAction，则前往 action 节点。
4. action 节点执行指定的工具，并将结果记录到 intermediate_steps。
5. 通过 workflow.add_edge("action", "agent") 这条边，执行完毕后自动回到 agent 节点。
6. agent 节点再次思考，此时它的状态里已经包含了上一步的工具执行结果（在 intermediate_steps 里），它可以基于这个新信息做出下一个决策。如此循环，直到它认为任务完成，输出 AgentFinish。

4.3 编译并运行你的智能Agent

图定义好了，最后一步是“编译”，把它变成一个可以调用的对象。

# 7. 编译图，得到可执行的应用
app = workflow.compile()

# 现在，你可以像调用函数一样运行这个Agent了！
# 准备初始输入状态
initial_state: AgentState = {
    "input": "北京现在的温度是多少？然后分析一下‘今天天气真好啊’这句话的情感和语言。",
    "chat_history": [], # 新对话，历史为空
    "agent_outcome": None, # 初始无决策
    "intermediate_steps": [] # 初始无执行步骤
}

# 运行Agent
final_state = app.invoke(initial_state)

# 查看最终结果
print("=== 运行结束 ===")
print("最终Agent输出:", final_state["agent_outcome"])
print("完整的执行步骤记录:", final_state["intermediate_steps"])

当你运行这段代码，你会看到控制台输出Agent一步步的思考和行为。它可能会先调用天气工具查询北京温度，然后在 intermediate_steps 里有了这个结果后，下一轮思考再调用情感分析工具处理第二个任务。整个过程完全自动化，无需你手动干预流程。

5. 进阶技巧与踩坑心得

一个基础的、能跑通的Agent构建完成了。但想让它真正可靠、实用，还需要注意以下几点，这些都是我在项目里真金白银踩出来的坑。

5.1 如何调试你的LangGraph应用？

当Agent行为不符合预期时，别急着改代码，先看清楚它到底做了什么。LangGraph提供了非常棒的调试支持。

方法一：可视化你的图。

# 将图导出为PNG图片
from langgraph.graph import Graph
png_data = workflow.get_graph().draw_mermaid_png()
with open("my_agent_graph.png", "wb") as f:
    f.write(png_data)

生成一张流程图，能帮你直观确认节点和边的连接关系是否正确。

方法二：开启详细日志，追踪状态变化。 LangGraph内部使用Pydantic管理状态。你可以通过设置环境变量来查看详细的状态转换日志，这对于理解 intermediate_steps 是如何被 operator.add 更新的尤其有用。不过更直接的是在你自己的节点函数里多打 print，就像我在 execute_tools_node 里做的那样。

方法三：分步执行（Stepping Through）。 app.compile() 返回的对象有一个 .stream() 方法，可以让你迭代地执行每一步，观察中间状态。

inputs = {"input": "你好", "chat_history": [], "agent_outcome": None, "intermediate_steps": []}
for step in app.stream(inputs, stream_mode="values"):
    print(f"步骤输出: {step}")

这能让你像调试器一样，看清楚每一轮循环后状态的具体变化。

5.2 处理复杂分支与多Agent协作

我们上面构建的是单Agent循环图。LangGraph的强大之处在于能轻松构建更复杂的拓扑结构。

• 并行节点：你可以定义多个工具执行节点，然后通过条件边，让Agent决定调用哪一个。这需要你在 should_continue 函数里返回更多的分支键（如 "call_tool_a", "call_tool_b"），并在 add_conditional_edges 的映射字典里配置好。
• 多Agent协作（Supervisor）：你可以创建多个不同的 agent_runnable（比如一个擅长分析，一个擅长总结），然后设计一个“主管”节点或路由逻辑，根据问题类型将任务分发给不同的专家Agent。这本质上就是创建了多个“思考”节点，并用更复杂的条件边或普通边将它们连接起来。
• 子图（Subgraph）：你可以将一个复杂的子流程（例如一个完整的数据库查询-分析流程）封装成一个单独的 StateGraph，然后通过 add_node 将其作为一个大节点加入到主图中。这极大地提升了模块化和复用性。

5.3 提升工具调用的准确性与稳定性

工具调用不准，多半是提示（Prompt）和工具描述的问题。

• 优化Prompt：hub.pull("hwchase17/openai-tools-agent") 是一个很好的起点，但针对你的具体任务，可能需要微调。特别是要在Prompt里明确告诉Agent：“你必须严格按照给定的工具格式调用，不能自己编造参数”。
• 细化工具描述和参数描述：回顾第2.1节，这是最重要的。在 Field(description=...) 里，用最清晰无歧义的语言描述这个参数是什么、格式要求（是字符串还是数字、有没有单位、有没有示例）。大模型对这部分文本的理解直接决定了它能否正确填充参数。
• 处理失败情况：我们的 get_current_temperature 工具里有一个简单的错误处理。但在生产环境中，你需要在 execute_tools_node 中考虑更健壮的异常捕获。当工具执行失败时，是重试、换一个工具，还是将错误信息作为 observation 返回给Agent让它决定？这都需要设计。

我在实际项目中，还喜欢为重要的工具编写单元测试，用一些典型的用户问题去测试Agent是否能正确触发工具并传参。这能有效避免因描述不清导致的“幻觉调用”。

构建基于LangGraph的Agent，一开始可能会觉得概念有点多，但一旦你理解了“状态流转”和“图结构”这两个核心，就会发现它带来的清晰度和可控性是巨大的。它迫使你以结构化的方式思考智能体的工作流，这对于开发复杂、可靠的AI应用来说，不是负担，而是最好的助力。从今天这个能调用三个工具的简单Agent开始，试着给它增加新功能，设计更巧妙的流程，你会发现，用代码“绘制”智能体的行为，是一件非常有成就感的事。