
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="你是一个帮助用户查询海洋天气的智能助手。" "以下是查询海洋天气信息的指南:" "" + " " ) ] ), 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()如果要查询加利福尼亚州圣克莱门特(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。 " + "
评论(0)