深入解析 LangGraph 智能体开发工作流：从概念到实践

Manus 再次让我们看到 LLM ReAct （Reasoning Act，推理+行动）模式的强大能力。其实，我们也可以自己通过 LangGraph 提供的智能体（Agent）构建工作流实现相似的功能。让 LLM（大语言模型）为我们完成更加复杂和实用的功能。

著名 AI 科普达人 New Machina，展示了如何使用 LangGraph 来搭建一个基于 ReAct 范式的智能体系统，该系统能够调用国家气象服务浮标 API，获取实时海洋天气数据，并将结果返回给用户。

整个流程的核心在于使用图数据结构（Graph Data Structure）来组织智能体的决策路径。每个节点代表一个状态或计算步骤，而边则定义了不同状态之间的转换逻辑。在 Python 代码的实现中，LangGraph 允许开发者以简洁的方式定义这些节点和边，使得整个工作流更加直观和易于管理。这种方式的好处在于，它不仅让智能体的行为逻辑更清晰，而且极大增强了系统的可扩展性。如果开发者需要增加更多工具或决策步骤，只需调整图的结构，而不必推倒重来。

在具体实现上，该智能体工作流的核心逻辑围绕 ReAct 模型展开。用户输入一个查询，比如 “圣克莱门特海岸附近的浪高是多少？”，LLM 首先进行推理，判断自身是否具备足够的知识来直接回答问题。如果发现训练数据不足，它会决定调用一个外部工具，例如浮标信息 API，来获取实时信息。这个过程中，LLM 还会推断出应该使用哪个浮标 ID，比如对于圣克莱门特，它会选择 46086 号浮标，而对于半月湾，则会选择 46214 号浮标。

这一点非常关键，因为它展示了 LLM 在一个结构化环境中的 “智能决策能力”。传统的 LLM 主要依赖自身的训练数据进行回答，而在 ReAct 工作流中，模型可以主动调用外部 API，填补知识空白，这使其在特定领域的应用能力大幅提升。例如，在金融、医疗、供应链等领域，类似的方法可以用于实时数据查询，提高 LLM 作为智能助手的实用性。

此外，New Machina 提到的 LangGraph 工具绑定机制也是一个亮点。通过 Python 装饰器 @tool，开发者可以将外部 API 直接注册到智能体系统中，使得 LLM 可以在推理过程中自然地调用这些工具。这种方式不仅简化了开发流程，还赋予了系统更高的透明性和可维护性。比如，如果未来国家气象服务 API 发生变化，开发者只需要修改工具函数，而不需要调整整个智能体逻辑。

从更宏观的角度来看，LangGraph 代表了一种新的 AI 应用范式。过去，我们主要依赖 LLM 进行 “静态推理”，也就是基于已有知识回答问题。而 ReAct 工作流的引入，则让智能体具备了“动态推理” 和 “主动执行” 的能力。这不仅扩展了 LLM 的应用边界，也让 AI 系统能够更加自然地融入实际业务流程。

当然，这种方法也有其挑战。比如，如何确保 LLM 在调用工具时不会误用 API？如何在多工具环境下优化决策逻辑？如何在不同任务之间高效管理智能体的状态？这些问题都值得进一步探索。但可以肯定的是，结合推理和行动的智能体将成为未来 AI 发展的重要方向，而 LangGraph 正在为这个未来提供一条清晰的路径。

完整示例代码

import certifi
import urllib3
import csv
from langchain_core.tools import tool
from typing_extensions import Literal
from langchain_openai import ChatOpenAI
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph import MessagesState
from langchain_core.messages import SystemMessage, HumanMessage, ToolMessage

class MarineForecast:
    def __init__(self):
        self.wave_height = 0  # 波高（单位：英尺）
        self.wave_period = 0  # 波浪周期（单位：秒）
        return

    def getHumanReadableStr(self):
        # 返回可读的天气信息
        return "The waves are " + str(self.wave_height) + " feet with period of " + str(self.wave_period) + " seconds."

# 初始化 LLM（GPT-4）
llm = ChatOpenAI(model="gpt-4", temperature=0)

@tool
def getMarineForecast(buoyId: str) -> str:
    """
    调用美国国家气象局（NWS）获取指定浮标（buoyId）附近的海洋天气预报。

    参数:
        buoyId (str): 浮标 ID，用于查询海洋天气信息。

    返回值:
        MarineForecast: 包含波高和周期信息的字符串。
    """
    # 获取海洋天气浮标数据
    http = urllib3.PoolManager(cert_reqs='CERT_REQUIRED', ca_certs=certifi.where())
    response = http.request('GET', 'https://www.ndbc.noaa.gov/data/realtime2/' + buoyId + '.txt')
    lines = response.data.decode("utf-8").splitlines()
    
    reader = csv.reader(lines)
    next(reader, None)  # 跳过第一行表头
    next(reader, None)  # 跳过第二行表头

    marine_forecast = MarineForecast()
    
    for row in reader:
        rowDatum = row[0]
        rowList = rowDatum.split()
        
        if rowList and rowList[8] != 'MM' and rowList[9] != 'MM':  # 确保数据有效
            marine_forecast.wave_height = round(float(rowList[8]) * 3.28084, 1)  # 转换单位（米 -> 英尺）
            marine_forecast.wave_period = rowList[9]
            break

    return marine_forecast.getHumanReadableStr()

# 为 LLM 增强工具功能
tools = [getMarineForecast]
tools_by_name = {tool.name: tool for tool in tools}
llm_with_tools = llm.bind_tools(tools)

# 定义对话节点
def llm_call(state: MessagesState):
    """LLM 处理用户请求"""

    return {
        "messages": [
            llm_with_tools.invoke(
                [
                    SystemMessage(
                        content="你是一个帮助用户查询海洋天气的智能助手。"
                                "以下是查询海洋天气信息的指南："
                                "" +
                                "如果要查询加利福尼亚州圣克莱门特（San Clemente）的海洋天气，请检查浮标 ID 46086。" +
                                "如果要查询加利福尼亚州康塞普西翁角（Point Conception）的海洋天气，请检查浮标 ID 46054。" +
                                "如果要查询加利福尼亚州使命湾（Mission Bay）的海洋天气，请检查浮标 ID 46258。" +
                                "如果要查询加利福尼亚州洛马角（Point Loma）的海洋天气，请检查浮标 ID 46232。" +
                                "如果要查询加利福尼亚州圣塔莫尼卡湾（Santa Monica Bay）的海洋天气，请检查浮标 ID 46221。" +
                                "如果要查询加利福尼亚州半月湾（Half Moon Bay）的海洋天气，请检查浮标 ID 46214。" +
                                "如果要查询加利福尼亚州 Mavericks 的海洋天气，请检查浮标 ID 46214。" +
                                ""
                    )
                ]
            ),
            state["messages"]
        ]
    }

def tool_node(state: dict):
    """执行 LLM 生成的工具调用"""
    
    result = []
    
    for tool_call in state["messages"][-1].tool_calls:
        tool = tools_by_name[tool_call["name"]]
        observation = tool.invoke(tool_call["args"])
        result.append(ToolMessage(content=observation, tool_call_id=tool_call["id"]))
    
    return {"messages": result}

# 判断是否需要调用工具
def should_continue(state: MessagesState) -> Literal["environment", END]:
    """决定是否继续调用工具或结束对话"""
    
    messages = state["messages"]
    last_message = messages[-1]
    
    # 如果 LLM 生成了工具调用，则执行工具调用
    if last_message.tool_calls:
        return "Action"
    
    # 否则，直接结束对话（回复用户）
    return END

# 构建对话流程
agent_builder = StateGraph(MessagesState)

# 添加节点
agent_builder.add_node("llm_call", llm_call)
agent_builder.add_node("environment", tool_node)

# 连接节点
agent_builder.add_edge(START, "llm_call")
agent_builder.add_conditional_edges(
    "llm_call",
    should_continue,
    {
        # should_continue 的返回值 -> 下一个要访问的节点
        "Action": "environment",
        END: END,
    },
)

agent_builder.add_edge("environment", "llm_call")

# 编译智能代理
agent = agent_builder.compile()

# 运行查询示例
# messages = [HumanMessage(content="Mavericks in Half Moon Bay 的波高和周期是多少？")]
# messages = [HumanMessage(content="Santa Monica 的波高和周期是多少？")]
messages = [HumanMessage(content="San Clemente 的波高和周期是多少？")]

messages = agent.invoke({"messages": messages})

# 输出结果
for m in messages["messages"]:
    m.pretty_print()