https://langchain-ai.github.io/langgraph/
当已完成前面两篇的案例之后，对于LangGraph的Agent开发就有了比较直观的感受。它的最大特色就在于它的图设计的方法：把工具封成节点，再以边去做调用逻辑。
Agent开发已经没有神秘可言，接下来就是一些添加功能和封装的工作啰。
有了LangGraph的图设计和Tool Calling Agent打底，可以尝试ReAct Agent。
简介
ReAct（Reasoning and Action），通过思维链的方式，引导模型将复杂问题进行拆分，一步一步地进行推理（Reasoning）和行动（Action），同时还引入了观察（Observation）环节，在每次执行（Action）之后，都会先观察（Observation）当前现状，然后再进行下一步的推理（Reason）。ReAct这个框架，就是要让LLM，进行推理，然后采取行动与外界环境互动。
当有了工具列表和模型后，就可以通过create_react_agent这个LangGraph框架中预构建的方法来创建自治循环代理（ReAct）的工作流，其必要的参数如下：
- model： 支持工具调用的LangChain聊天模型。
- tools： 工具列表、ToolExecutor 或 ToolNode 实例。
- state_schema：图的状态模式。必须有
messages和is_last_step键。默认为定义这两个键的Agent State。
通过对不同复杂程度输入问题的测试，我们发现当前架构能够非常准确且快速地完成任务目标。在涉及多个任务的顺序执行时，ReAct 代理能够自主决策并执行，真正实现了完全的自治循环代理。此外，其可扩展性也十分出色。对于不同的业务需求，我们只需调整接入的大模型实例（可使用其他开源或在线模型）作为 ReAct 的基础模型。对于工具的配置，也无需特别进行复杂的编排，只需明确定义每个工具的输入和输出，然后通过工具列表的形式直接注册到大模型实例及 ToolNode 实例中。这种方法在快速构建智能代理方面，非常值得大家尝试。
完整案例
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) ,确实比之前两个案例简单直接了很多。
从案例的运行中可以看到，当有大模型不知道时就会调用外部工具来帮忙解决问题，直到解决为止。聪明如你，一定会想到，可以照此多加几个工具就可让这个Agent的能力上天入地啊！
流式输出
流式输出功能在LangGraph 框架中的实现方式比较简单，因为LangGraph底层是基于 LangChain 构建的，所有就直接把LangChain中的回调系统拿过来使用了。在LangChain中的流式输出是：以块的形式传输最终输出，即一旦监测到有可用的块，就直接生成它。最常见和最关键的流数据是大模型本身生成的输出。 大模型通常需要时间才能生成完整的响应，通过实时流式传输输出，用户可以在生成时看到部分结果，这可以提供即时反馈并有助于减少用户的等待时间。
LangGraph框架中的工作流中由各个步骤的节点和边组成。这里的流式传输涉及在各个节点请求更新时跟踪图状态的变化。这样可以更精细地监控工作流中当前处于活动状态的节点，并在工作流经过不同阶段时提供有关工作流状态的实时更新。其实现方式也是和LangChain一样通过.stream和.astream方法执行流式输出，只不过适配到了图结构中。调用.stream和.astream方法时可以指定几种不同的模式，即：
- "values" ：在图中的每个步骤之后流式传输状态的完整值。
- "updates" ：在图中的每个步骤之后将更新流式传输到状态。如果在同一步骤中进行多个更新（例如运行多个节点），则这些更新将单独流式传输。
- "debug" ：在整个图的执行过程中流式传输尽可能多的信息，主要用于调试程序。
- "messages"：记录每个
messages中的增量token。 - "custom"：自定义流，通过
LangGraph 的 StreamWriter方法
async def main():
async for event in graph.astream_events({"messages": ["what is labubu"]}):
kind = event["event"]
if kind == "on_chat_model_stream":
print(event["data"]["chunk"].content, flush=True)
asyncio.run(main())
需要流式输出时，只需将最后的函数改成上面的方法即可。至此，我们就完整实现了在LangGraph中ReAct自治代理的完整构建。