Agent开发，ReAct Agent / ai #45

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自治代理的完整构建。