当有了工具列表和模型后,就可以通过create_react_agent这个LangGraph框架中预构建的方法来创建自治循环代理(ReAct)的工作流,其必要的参数如下:
messages和is_last_step键。默认为定义这两个键的Agent State。from dotenv import dotenv_values
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from typing import Union, Optional
from pydantic import BaseModel, Field
import requests, json, asyncio
from langgraph.prebuilt import create_react_agent
env_vars = dotenv_values('.env')
OPENAI_KEY = env_vars['OPENAI_API_KEY']
OPENAI_BASE_URL = env_vars['OPENAI_API_BASE']
SERPER_KEY = env_vars['SERPER_KEY']
WEATHER_KEY = env_vars['WEATHER_KEY']
## 第一个工具
class WeatherLoc(BaseModel):
location: str = Field(description="The location name of the city")
@tool(args_schema=WeatherLoc)
def get_weather(location):
"""
Function to query current weather.
:param loc: Required parameter, of type string, representing the specific city name for the weather query. \
Note that for cities in China, the corresponding English city name should be used. For example, to query the weather for Beijing, \
the loc parameter should be input as 'Beijing'.
:return: The result of the OpenWeather API query for current weather, with the specific URL request address being: https://api.openweathermap.org/data/2.5/weather. \
The return type is a JSON-formatted object after parsing, represented as a string, containing all important weather information.
"""
# Step 1.构建请求
url = "https://api.openweathermap.org/data/2.5/weather"
# Step 2.设置查询参数
params = {
"q": location,
"appid": WEATHER_KEY,
"units": "metric",
"lang":"zh_cn"
}
# Step 3.发送GET请求
response = requests.get(url, params=params)
# Step 4.解析响应
data = response.json()
return json.dumps(data)
# 第二个工具
class SearchQuery(BaseModel):
query: str = Field(description="Questions for networking queries")
@tool(args_schema = SearchQuery)
def fetch_real_time_info(query):
"""Get real-time Internet information"""
url = "https://google.serper.dev/search"
payload = json.dumps({
"q": query,
"num": 1,
})
headers = {
'X-API-KEY': SERPER_KEY,
'Content-Type': 'application/json'
}
response = requests.post(url, headers=headers, data=payload)
data = json.loads(response.text)
if 'organic' in data:
return json.dumps(data['organic'], ensure_ascii=False)
else:
return json.dumps({"error": "No organic results found"}, ensure_ascii=False)
llm = ChatOpenAI(model="gpt-4o-mini", api_key=OPENAI_KEY,base_url=OPENAI_BASE_URL)
tools = [fetch_real_time_info, get_weather]
graph = create_react_agent(llm, tools=tools)
# 可以自动处理成 HumanMessage 的消息格式
finan_response = graph.invoke({"messages":["what is labubu"]})
print(569, finan_response)
# finan_response["messages"][-1].content
以上案例中使用了两个外部工具[fetch_real_time_info, get_weather], 图的生成也只用了一行代码create_react_agent(llm, tools=tools) ,确实比之前两个案例简单直接了很多。
RE: LangGraph开发实战