ReAct в LangGraph: от ручно написанного цикла к ToolNode и StateGraph

Базовый цикл вызова инструментов выполняется в один проход: модель запрашивает инструменты, приходят результаты, затем отправляется ответ. Задачи, требующие последовательных вызовов, когда второй вызов зависит от результата первого, нуждаются в агенте, способном мыслить, действовать, оценивать результат и снова действовать. Этот цикл и является шаблоном ReAct. Ниже показана суть, ручно написанный цикл ReAct, а также тот же агент, переписанный с использованием LangGraph с узлом инструмента, сообщениями об инструментах, условным ребром и графом состояний.

Элементы, предоставляемые LangGraph для этого:

• Узел инструмента: парсит запросы на использование инструментов, запускает их и возвращает результаты.
• Сообщение об инструменте: передает результаты работы инструмента обратно в LLM.
• tools_condition: функция маршрутизации, которая проверяет наличие запросов к инструментам и управляет условным ребром.
• StateGraph: запускает рабочий процесс.

Состояние: словарь с введёнными данными, хранящий изменяющиеся параметры агента.

Узлы: функции на Python, обновляющие состояние; одним из таких узлов является узел инструмента.

Рёбра: переходы между узлами, причём условные рёбра позволяют создавать ветвления и циклы.

Настройка

Храните конфиденциальные данные в файле .env в корне проекта:

OPENAI_API_KEY="<Your OpenAI API Key>"
TAVILY_API_KEY=<Your TAVILY API Key>

Загрузите их с помощью класса BaseSettings из Pydantic:

class BaseConfig(BaseSettings):
    OPENAI_API_KEY: Optional[str]
    PINECONE_API_KEY: Optional[str]
    TAVILY_API_KEY: Optional[str]

model_config = SettingsConfigDict(env_file=".env", extra="ignore")

Структура проекта:

project/
├── react_loop/
│   └── weather_assistant.py # implement a vanilla react loop
│
├── react_langgraph     # implement ReAct by LangGraph workflow
│   ├── nodes.py        # graph nodes. Defined as Python functions
│   ├── graph.py        # graph workflow
│   ├── consts.py       # define node names as constants
│   └── state.py        # typed dictionary to store graph states
|
├── tools
│   ├── tool_call.py  # implement simple tool-calling loop
|   └── weather.py    # get_weather LangChain tool using open source APIs
|
|──    # implement tool-calling loop using LangChain tool
├── config.py         # pydantic BaseConfig
├── utils.py          # utility method to print messages
├── .env              # environment variable definition
├── .gitignore
└── requirements.txt  # package requirements

Что означает ReAct

ReAct — это аббревиатура от Reasoning plus Acting. Агент, выполняющий многоэтапную задачу, рассуждает, действует, наблюдает и снова рассуждает до её завершения. Базовый цикл позволяет выполнить только один заход инструментов, и именно этот недостаток устраняет ReAct.

Один цикл включает шесть этапов:

1. Пользователь отправляет сообщение.
2. LLM читает запрос, определяет необходимую информацию и решает, какие инструменты вызвать.
3. Агент запускает один или несколько инструментов, таких как поиск в интернете, запрос к API или отправка электронного письма.
4. Результаты возвращаются и добавляются к текущему состоянию агента.
5. LLM оценивает новую информацию и либо отвечает, либо запрашивает дополнительные действия с использованием инструментов.
6. Когда дальнейших действий не требуется, выводится окончательный ответ.

Цикл является мощным, поскольку агент может скорректировать свой план в соответствии с тем, что только что узнал, так что каждое решение может зависеть от предыдущих результатов.

Простой цикл ReAct

Эта версия находится в файле react_loop/weather_assistant.py и похожа на цикл с одним проходом, но имеет два отличия:

async def chat(self, message: str):
    # Add user message
    self.messages.append(HumanMessage(content=message))

    # loop to handle serial tool calls
    while True:

        # Get AI response with potential tool calls
        response = await self.llm_with_tools.ainvoke(self.messages)
        self.messages.append(response)

        # Check if there is any tools to call
        if not response.tool_calls:
            # no tools to call
            break
            # Execute each tool call

        # process tool calls asynchronously
        results = await asyncio.gather(*[self.tools[tc["name"]].ainvoke(tc) for tc in response.tool_calls])
        self.messages.extend(results)

Во-первых, вызовы инструментов выполняются асинхронно с помощью asyncio.gather, поэтому несколько из них могут работать параллельно. Во-вторых, шаг выполнения инструмента находится внутри цикла while, где постоянно сначала формируется логика, затем выполняется действие, пока в ответе модели больше не будут содержаться вызовов инструментов.

Тот же агент в LangGraph

Ручной цикл работает, но требует от вас личного управления циклом «логика — действие», списком сообщений, словарем инструментов и сбором результатов. LangGraph берет на себя каждый из этих аспектов: ToolNode парсит вызовы и выполняет инструменты, StateGraph хранит сообщения в состоянии графа, а ребро tools_condition заменяет ручной цикл while и связанный с ним проверочный код на вызовы инструментов.

Состояние. Словарь определенного типа хранит список сообщений (в файле react_langgraph/state.py):

from typing import TypedDict, Annotated
from langgraph.graph.message import AnyMessage, add_messages

class ReactState(TypedDict):
    """
    Define the state schema for our hotel assistant.

    Attributes:
        messages: list of messages
    """

    messages: Annotated[list[AnyMessage], add_messages]

Узлы считывают и обновляют сообщения через него, а функция-редюсер add_messages добавляет всё, что возвращает узел.

Узлы. В файле react_langgraph/nodes.py происходит создание списка инструментов LangChain, системного промпта, который указывает модели, как ими пользоваться, а также модели OpenAI, привязанной к этому списку инструментов:

tools = [TavilySearch(max_results=3, tavily_api_key=TAVILY_API_KEY), get_weather]

SYSTEM_PROMPT = f"""You are an assistant for question-answering tasks with access to multiple tools.
    use these tools to retrieve context to answer the question. If you don't know the answer,
    just say that you don't know. Use three sentences maximum and keep the answer concise.

    Tool Use Instructions:

    - Use tools thoughtfully—only when they add value to the response
    - Chain tools when needed (e.g., search for info, then use that to search
      for relevant hotels)
      - Be conversational and helpful
    """

system_message = SystemMessage(content=SYSTEM_PROMPT)

llm = ChatOpenAI(api_key=api_key, model="gpt-4o-mini", temperature=0)
llm_with_tools = llm.bind_tools(tools)


async def run_agent_reasoning(state: ReactState) -> dict:
    """
    run agent reasoning node
    :param state: ReactState, manage graph state
    :return: dict containing new message
    """

    print("----AGENT REASONING------")
    messages = [system_message] + state["messages"]

    if isinstance(messages[-1], ToolMessage):
        print_message(messages[-1])

    response = await llm_with_tools.ainvoke(messages)
    print_message(response)

    return {"messages": [response]}


tool_node = ToolNode(tools)

В результате получаются два узла. Узел рассуждений формирует список сообщений, вызывает LLM и возвращает его ответ. Узел действия представляет собой ToolNode, который парсит вызовы инструментов, запускает их и возвращает сообщения от инструментов.

Граф. В файле react_langgraph/graph.py объект StateGraph регистрирует узлы и соединяет их рёбрами:

workflow = StateGraph(ReactState)

workflow.add_node(AGENT_REASON, run_agent_reasoning)
workflow.add_node(ACT, tool_node)

workflow.add_edge(START, AGENT_REASON)
workflow.add_edge(ACT, AGENT_REASON)

workflow.add_conditional_edges(
    AGENT_REASON,
    tools_condition,
    {"tools": ACT, END: END}
)tion is invoked first

Выполнение происходит согласно четырем правилам. Граф начинается с узла рассуждений. Если ответ LLM содержит вызовы инструментов, tools_condition направляет поток к узлу действия. После выполнения инструментов управление всегда возвращается к рассуждениям. Если в ответе нет вызовов инструментов, граф завершается и возвращается окончательный ответ. Это соответствует циклу ReAct без ручного цикла или управляемых вручную сообщений.

Компилируйте и запускайте. Компиляция создает агента, а диаграмму графа можно экспортировать в формате PNG:

react_agent = workflow.compile()

react_agent.get_graph().draw_mermaid_png(output_file_path="graphs.png")

question = """
           tell me the first hotel you find in Boston if
            its temperature is lower than 30 degree."""


async def main():
    print("hello ReAct agent by Langraph")
    message = HumanMessage(content=question)
    res = await react_agent.ainvoke({"messages": [message]})


if __name__ == "__main__":
    asyncio.run(main())

Вызовите его с пользовательским сообщением, и полученный список сообщений покажет узел рассуждения и узлы инструментов, выполняющие свои действия, что представляет собой рабочий многоэтапный процесс.

Что это дает

Для выполнения задач используются три функции LangGraph: ToolNode для поиска и запуска инструментов, tools_condition для управления циклами и add_messages для автоматической обработки сообщений. После того как концепция ReAct будет понята и реализована, следующим шагом станет возвращение к пайплайну LangGraph RAG и превращение его в полностью агентного решения.