ReAct w LangGraph: Od ręcznie zbudowanej pętli do ToolNode i StateGraph

Bazowy cykl wywoływania narzędzi obejmuje jedną rundę: model prosi o narzędzia, przychodzą wyniki, a następnie wysyłana jest odpowiedź. Zadania wymagające łańcuchowych wywołań, w których drugie wywołanie zależy od tego, co zwróciło pierwsze, potrzebują agenta, który potrafi myśleć, działać, ocenić wynik i ponownie działać. Ten cykl to wzorzec ReAct. Poniżej znajduje się wyjaśnienie koncepcji, ręcznie napisany cykl ReAct oraz ten sam agent odbudowany w LangGraph z użyciem węzła narzędzia, wiadomości o narzędziach, warunkowego krawędzi oraz grafu stanów.

Części dostarczane przez LangGraph do tego celu:

• Węzeł narzędzia: analizuje wywołania narzędzi, uruchamia je i zwraca wyniki.
• Wiadomość o narzędziu: przekazuje wynik narzędzia z powrotem do LLM.
• tools_condition: funkcja routingu, która sprawdza wywołania narzędzi i steruje warunkowym krawędziem.
• StateGraph: uruchamia proces pracy.

Stan: słownik typowany, który przechowuje zmieniające się dane agenta.

Węzły: funkcje w Pythonie, które aktualizują stan, przy czym jeden z nich to węzeł narzędziowy.

Krawędzie: przejścia pomiędzy węzłami, przy czym krawędzie warunkowe umożliwiają tworzenie gałęzi i pętli.

Ustawienia

Zachowaj dane poufne w pliku .env znajdującym się w korzeniu projektu:

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

Ładuj je za pomocą klasy Pydantic BaseSettings:

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")

Rozkład projektu:

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

Co oznacza ReAct

ReAct to skrót od Reasoning plus Acting. Agent realizujący złożone zadanie kroczy etapami: rozumuje, działa, obserwuje i ponownie rozumuje, aż zadanie zostanie ukończone. Podstawowy cykl umożliwia tylko jedną rundę wykonywania narzędzi, co jest właśnie tym brakiem, który ReAct eliminuje.

Jedna runda obejmuje sześć kroków:

1. Użytkownik wysyła wiadomość.
2. LLM czyta prośbę, określa, jakie informacje są potrzebne, i decyduje, które narzędzia należy użyć.
3. Agent uruchamia jedno lub więcej narzędzi, takich jak wyszukiwanie w internecie, zapytanie do API lub wysłanie e-maila.
4. Wyniki wracają i są uwzględniane w aktualnym stanie agenta.
5. LLM ocenia nowe informacje i albo udziela odpowiedzi, albo prosi o dodatkowe wywołania narzędzi.
6. Gdy nie są już potrzebne żadne dalsze działania, wysyłana jest ostateczna odpowiedź.

Pętla ta jest skuteczna, ponieważ agent może dostosować swój plan na podstawie tego, czego właśnie się nauczył, dzięki czemu każda decyzja może opierać się na wcześniejszych wynikach.

Zwykła pętla ReAct

Ta wersja znajduje się w pliku react_loop/weather_assistant.py i przypomina pętlę jednopassową, z dwiema różnicami:

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)

Najpierw wywołania narzędzi są realizowane asynchronicznie za pomocą asyncio.gather, dzięki czemu kilka z nich może być wykonywanych równolegle. Po drugie, krok wykonania narzędzia znajduje się wewnątrz pętli while, która powtarza sekwencję „uzasadnienie – działanie” aż odpowiedź modelu nie będzie już zawierać wywołań narzędzi.

Ten sam agent w LangGraph

Ręczna pętla działa, ale zmusza do samodzielnego zarządzania cyklem uzasadnienie-działanie, listą wiadomości, słownikiem narzędzi oraz zbieraniem wyników. LangGraph przejmuje kontrolę nad tym wszystkim: ToolNode analizuje wywołania i wykonuje narzędzia, StateGraph przechowuje wiadomości w stanie grafu, a krawędź tools_condition zastępuje ręcznie napisaną pętlę while oraz jej sprawdzanie wywołań narzędzi.

Stan. Skatalogowany słownik przechowuje listę wiadomości (w pliku 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]

Węzły czytają i aktualizują wiadomości za jego pośrednictwem, a reduktor add_messages dodaje wszystko, co zwraca węzeł.

Węzły. W pliku react_langgraph/nodes.py konfiguracja tworzy listę narzędzi LangChain, prompt systemowy, który mówi modelowi, jak je używać, oraz model OpenAI powiązany z tą listą narzędzi:

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)

Powstają dwa węzły. Węzeł rozumowania tworzy listę wiadomości, wywołuje LLM i zwraca jego odpowiedź. Węzeł działania to ToolNode, który analizuje wywołania narzędzi, uruchamia je i zwraca wiadomości od tych narzędzi.

Graf. W pliku react_langgraph/graph.py StateGraph rejestruje węzły i łączy je krawędziami:

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

Wykonywanie odbywa się zgodnie z czterema regułami. Graf rozpoczyna się od węzła rozumowania. Jeśli odpowiedź LLM zawiera wywołania narzędzi, tools_condition kieruje przepływ do węzła działania. Po zakończeniu pracy narzędzi kontrola zawsze wraca do etapu rozumowania. Jeśli odpowiedź nie zawiera żadnych wywołań narzędzi, graf się kończy i jest zwracana ostateczna odpowiedź. To odzwierciedla cykl ReAct bez ręcznego pętli ani zarządzanych ręcznie wiadomości.

Kompiluj i uruchamiaj. Kompilacja tworzy agenta, a diagram grafu można eksportować jako plik 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())

Zawołaj to za pomocą wiadomości od użytkownika, a powstała lista wiadomości pokaże węzeł rozumowania oraz kolejne kroki węzła narzędzia, co stanowi działający wieloetapowy proces.

Jak to wygląda obecnie

Do realizacji tego zadania służą trzy funkcje LangGraph: ToolNode do znajdowania i uruchamiania narzędzi, tools_condition do sterowania pętlami oraz add_messages do automatycznego obsługi wiadomości. Gdy ReAct zostanie zrozumiany i wdrożony, kolejnym krokiem jest powrót do pipeline RAG w LangGraph i uczynienie go w pełni agentowym.