LangGraph 是一个用于构建有状态 LLM 应用的框架,因此非常适合构建 ReAct(推理和行动)智能体。
ReAct 智能体将 LLM 推理与行动执行相结合。它们会迭代思考、使用工具并根据观察结果采取行动,以实现用户目标,并动态调整其方法。这种模式在“ReAct:在语言模型中协同推理和行动” (2023) 中首次提出,旨在模仿人类般的灵活问题解决方式,而不是僵化的工作流。
LangGraph 提供了一个预构建的 ReAct 智能体 (
create_react_agent),
当您需要对 ReAct 实现进行更多控制和自定义时,它会大放异彩。本指南将向您展示一个简化版本。
LangGraph 使用三个关键组件将智能体建模为图:
State:共享数据结构(通常为TypedDict或Pydantic BaseModel),表示应用的当前快照。Nodes:对智能体的逻辑进行编码。它们接收当前状态作为输入,执行一些计算或副作用,并返回更新后的状态,例如 LLM 调用或工具调用。Edges:根据当前State定义要执行的下一个Node,从而实现条件逻辑和固定转换。
如果您还没有 API 密钥,可以从 Google AI Studio 获取一个。
pip install langgraph langchain-google-genai geopy requests
在环境变量 GEMINI_API_KEY 中设置您的 API 密钥。
import os
# Read your API key from the environment variable or set it manually
api_key = os.getenv("GEMINI_API_KEY")
为了更好地了解如何使用 LangGraph 实现 ReAct 智能体,本指南将介绍一个实际示例。您将创建一个智能体,其目标是使用工具查找指定位置的当前天气。
对于此天气智能体,State 将维护正在进行的对话历史记录(作为消息列表)和一个计数器(作为整数),用于说明已采取的步骤数。
LangGraph 提供了一个辅助函数 add_messages,用于更新状态消息列表。它充当 reducer,
接收当前列表以及新消息,并返回合并后的列表。它通过消息 ID 处理更新,并默认为新消息和未见消息采用“仅追加”行为。
from typing import Annotated,Sequence, TypedDict
from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages # helper function to add messages to the state
class AgentState(TypedDict):
"""The state of the agent."""
messages: Annotated[Sequence[BaseMessage], add_messages]
number_of_steps: int
接下来,定义您的天气工具。
from langchain_core.tools import tool
from geopy.geocoders import Nominatim
from pydantic import BaseModel, Field
import requests
geolocator = Nominatim(user_agent="weather-app")
class SearchInput(BaseModel):
location:str = Field(description="The city and state, e.g., San Francisco")
date:str = Field(description="the forecasting date for when to get the weather format (yyyy-mm-dd)")
@tool("get_weather_forecast", args_schema=SearchInput, return_direct=True)
def get_weather_forecast(location: str, date: str):
"""Retrieves the weather using Open-Meteo API.
Takes a given location (city) and a date (yyyy-mm-dd).
Returns:
A dict with the time and temperature for each hour.
"""
# Note that Colab may experience rate limiting on this service. If this
# happens, use a machine to which you have exclusive access.
location = geolocator.geocode(location)
if location:
try:
response = requests.get(f"https://api.open-meteo.com/v1/forecast?latitude={location.latitude}&longitude={location.longitude}&hourly=temperature_2m&start_date={date}&end_date={date}")
data = response.json()
return dict(