ReAct Agent (with LangGraph)

ReAct Agent

ReAct는 LLM 기반 Agent 설계에서 주목받는 프레임워크 중 하나로, 추론과 행동을 긴밀하게 통합하는 독특한 접근 방식을 취한다.

https://arxiv.org/pdf/2210.03629

 

ReAct의 핵심 동기는 LLM 연구에서 개별적으로 발전해 온 두 가지 주요 능력, 즉 Chain-of-Thought (CoT)와 같은 추론 능력과 WebGPT, SayCan 등에서 보여준 행동(도구 사용, 환경 상호작용) 능력을 시너지 효과를 내도록 결합하는 것이다. 이는 인간이 계획 수립, 진행 상황 추적, 예외 처리 등을 위해 언어적 추론(내적 발화)과 실제 작업 수행을 자연스럽게 결합하는 방식과 유사하다.

 

핵심 메커니즘: 추론과 행동의 교차 수행

ReAct 프레임워크는 LLM에게 추론 흔적(사고 과정)과 작업 관련 행동을 번갈아 생성하도록 프롬프트한다. 이 과정은 "생각(Think)-행동(Act)-관찰(Observe)" 루프로 설명될 수 있다.

• 생각 (Think): Agent는 현재 상황(초기 작업, 이전 생각/행동/관찰 결과 등)을 바탕으로 다음에 무엇을 해야 할지, 어떤 정보를 찾아야 할지, 계획을 어떻게 수정해야 할지 등에 대해 자연어로 된 추론 흔적(thought)을 생성한다.
• 행동 (Act): 생성된 추론을 바탕으로 Agent는 구체적인 행동(action)을 결정하고 생성한다. 이 행동은 외부 도구(예: 검색 API 호출, 계산기 사용)를 사용하거나 환경과 상호작용하는 것일 수 있다.
• 관찰 (Observe): 행동이 실행되면, Agent는 그 결과(예: API 응답, 환경의 변화)를 관찰(observation)한다.
• 반복: 관찰 결과는 다음 "생각" 단계의 입력 컨텍스트에 포함되어, Agent가 상황을 업데이트하고 다음 추론과 행동을 계획하는 데 사용된다. 이 루프는 작업이 완료될 때까지 반복된다. 

 

(left)ReAct Agent의 기본 구조 (right)Langgraph 로 구현한 ReAct Agnet 기본 구조

ReAct는 추론이 행동을 이끌고(동적 계획 수립, 목표 분해), 행동이 다시 추론에 정보를 제공하는(환경/도구로부터 얻은 관찰 결과 통합) 양방향 시너지를 창출한다. 지식 집약적 추론 작업에서는 생각-행동-관찰 단계가 촘촘하게 반복될 수 있으며, 의사결정 작업에서는 행동 중간 중간 필요한 시점에 추론이 더 드물게 나타날 수 있다.

 

ReAct의 장점

• 향상된 사실 기반 추론 및 환각 감소: 외부 도구(예: 위키백과 API)와의 상호작용을 통해 추론 과정을 외부 정보에 기반(grounding)하게 함으로써, LLM의 내부 지식에만 의존하는 CoT 방식에서 발생하기 쉬운 사실 오류나 환각(hallucination) 문제를 완화한다.'
• 동적 계획 및 예외 처리: 교차 수행되는 추론을 통해 환경으로부터의 실시간 관찰 결과에 따라 계획을 동적으로 추적, 업데이트, 수정할 수 있으며, 예상치 못한 결과나 예외 상황에 효과적으로 대처할 수 있다.
• 해석 가능성 및 신뢰성 향상: 명시적으로 생성되는 추론 흔적은 Agent의 의사결정 과정을 투명하게 보여주어 인간이 이해하고 진단하기 쉽게 만든다.

---

 

중 에이전트 시스템(MAS): 산업 복잡성 해결

다중 에이전트 시스템(MAS)은 여러 자율 Agent가 상호작용하여 개별 Agent의 능력을 넘어서는 문제를 해결하는 시스템이다. 이는 인간 사회의 팀워크와 전문화 원리를 모방하여 , 복잡한 산업 환경의 문제를 효과적으로 해결하는 패러다임을 제공한다.

 

https://arxiv.org/pdf/2504.04650v2

 

Zihao Wu 는 Autono: A ReAct-Based Highly Robust Automomous Agent Framework 에서 ReAct 기반의 MAS 구현 방향을 제시했다.

ReAct Agent의 강력한 Tool Call 기능을 강화하여 HandoffTool 노드를 추가하였다. 이는 각각의 도메인에 특화된 Agnet가 유기적인 의사 결정을 하도록 돕는다.

 

MAS는 복잡한 산업 과제(예: 제조 계획, 공급망 최적화)를 더 작고 관리 가능한 하위 작업으로 분해하고, 이를 전문화된 Agent(예: 계획 Agent, 일정 관리 Agent, 자원 할당 Agent, 기계 Agent)에 할당하는 방식으로 작동한다.

 

 

 

import sys

sys.path.append('/root/LangGraph_KTL')

import importlib
import utils.nodes
import utils.state
importlib.reload(utils.nodes)
importlib.reload(utils.state)
importlib.reload(utils.tools)


from functools import partial
from pprint import pprint

from contextlib import asynccontextmanager
from langchain_core.messages import HumanMessage, AIMessage
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, END

from utils.tools import tools
from utils.nodes import get_tool_node, should_continue, call_model
from utils.state import AgentState, GraphConfig

import os
from pathlib import Path

from langchain_mcp_adapters.client import MultiServerMCPClient



@asynccontextmanager
async def make_graph():
    """Test that the MultiServerMCPClient can connect to multiple servers and load tools."""

    time_server_path = os.path.join("utils/servers/time_server.py")

    mcp_client =  MultiServerMCPClient(
        {
            "Local_time":{
                "command":"python",
                "args": [time_server_path],
                "transport": "stdio",
            },
            "desktop-commander": {
                "command": "npx",
                "args": [
                "-y",
                "@smithery/cli@latest",
                "run",
                "@wonderwhy-er/desktop-commander",
                "--key",
                "e538a05f-f1fb-446f-9138-c9fb115eb82a"
                ]
            },
        }
    )



    async with mcp_client as client:
        mcp_tools = client.get_tools()

        all_tools = mcp_tools + tools

        graph_builder = StateGraph(AgentState, config_schema=GraphConfig)

        graph_builder.add_node("chatbot", partial(call_model, tools=all_tools))

        all_tools = get_tool_node(all_tools)

        graph_builder.add_node("action", all_tools)

        graph_builder.add_conditional_edges(
            "chatbot",
            should_continue,{
                # If `tools`, then we call the tool node.
                "continue": "action",
                # Otherwise we finish.
                "end": END,
            }
        )
        # Any time a tool is called, we return to the chatbot to decide the next step
        graph_builder.add_edge("action", "chatbot")
        graph_builder.set_entry_point("chatbot")
        memory = MemorySaver()
        graph = graph_builder.compile()
        yield graph

async with make_graph() as graph:
    response = await graph.ainvoke({
        "messages": [HumanMessage(content="현재 파일경로에 오늘 날씨를 검색해서 {오늘 날짜}_날씨.html로 담아서 생성해줘.")]
    })
    pprint(response)