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()