Создание агента ReAct с нуля с использованием Gemini и LangGraph.

LangGraph — это фреймворк для создания приложений LLM с сохранением состояния, что делает его хорошим выбором для построения агентов ReAct (рассуждения и действия).

Агенты ReAct сочетают рассуждения на уровне языковых моделей с выполнением действий. Они итеративно мыслят, используют инструменты и действуют на основе наблюдений для достижения целей пользователя, динамически адаптируя свой подход. Этот шаблон, представленный в статье "ReAct: Synergizing Reasoning and Acting in Language Models" (2023), стремится имитировать человекоподобное, гибкое решение проблем, а не жесткие рабочие процессы.

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

В LangGraph агенты моделируются как графы с использованием трех ключевых компонентов:

  • State : Общая структура данных (обычно TypedDict или Pydantic BaseModel ), представляющая текущий снимок состояния приложения.
  • Nodes : Кодируют логику ваших агентов. Они получают текущее состояние в качестве входных данных, выполняют некоторые вычисления или побочные эффекты и возвращают обновленное состояние, например, вызовы LLM или вызовы инструментов.
  • Edges : Определяют следующий Node для выполнения на основе текущего State , допуская условную логику и фиксированные переходы.

Если у вас еще нет ключа API, вы можете получить его в Google AI Studio .

pip install langgraph langchain-google-genai geopy requests

Укажите свой API-ключ в переменной среды GEMINI_API_KEY .

import os

# Read your API key from the environment variable or set it manually
api_key = os.getenv("GEMINI_API_KEY")

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

Для этого метеорологического агента State будет вести историю текущих разговоров (в виде списка сообщений) и счетчик (в виде целого числа) количества пройденных шагов, в иллюстративных целях.

LangGraph предоставляет вспомогательную функцию add_messages для обновления списков сообщений в состоянии. Она функционирует как редуктор , принимая текущий список плюс новые сообщения и возвращая объединенный список. Она обрабатывает обновления по идентификатору сообщения и по умолчанию использует поведение "только добавление" для новых, ранее не встречавшихся сообщений.

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(zip(data[