[LangChain] AgentExecutor와 ReAct

AgentExecutor는 While Loop!

 

 랭체인이 제공하는 모듈인 에이전트는 일종의 동적 Chain으로 ReAct 기반 프롬프트로부터 action을 정하고 tool을 활용하며 상호작용할 수 있도록 한다. 예를 들어, 주어진 문자열의 길이를 반환하는 함수를 Tool로 사용하는 Agent를 만들고, 수행하면 다음과 같다.

# zero-shot ReAct Agent example
@tool
def get_text_length(text: str) -> int:
    """주어진 문자열의 길이를 반환하는 함수"""
    text = text.strip("'\n")
    return len(text)


if __name__ == "__main__":
    llm = ChatOpenAI()
    prompt = hub.pull("hwchase17/react")
    tools = [get_text_length]
    agent = create_react_agent(llm, tools, prompt)
    agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

    agent_executor.invoke({"input": "'ReAct'의 길이는?"})

> Entering new AgentExecutor chain...
문자열의 길이를 알아야 한다.
Action: get_text_length
Action Input: 'ReAct'
Observation: 5
Thought: 'ReAct'는 5글자로 이루어져 있다.
Final Answer: 5

> Finished chain.

 

 결과를 보면, AgnetExecutor 객체를 초기화하고, invoke 메소드를 통해 질의하면, Thought → Action → Action Input 을 도출하고 있다. 그럼 Observation은 Tool을 사용해야할텐데 어떻게 Tool을 호출하는지, 그리고 이를 기반으로 반복을 멈추는 것은 어떤 식으로 이루어지는지 알아보자.

---

AgentExecutor의 이해를 위해 pseudo-code를 살펴보자.

class FakeAgentExecutor:
    def invoke(self, input):
        while True:
            result = self.agent(input)
            if result == "RunTool":
                tool_to_run(tool_input)
            else:
                return result

 

 위 코드를 통해 AgentExecutor의 내부 동작 원리를 이해해보자.

1. 사용자 입력(input: What is the capital of Korea?)을 입력 받는다.
2. 입력과 Tool에 대한 정보, ReAct 등의 정보를 담은 prompt를 LLM에게 전달
3. LLM에게 답변을 받음(result)
   1. 만약 Tool을 실행해야 한다는 답변이라면, 어떤 Tool을 실행할지 알아내서 Tool 실행
   2. LLM이 최종 답변이라고 판단했을 때 혹은 max_iter 등에 도달했을 때 종료

Agent의 작동 원리

---

Agent from scratch

이번에는 문자열이 주어지면 길이를 반환하는 함수를 Tool로 정의하고 이를 사용하는 Agent를 만들어보자.

 

Tool 정의하기

 먼저 아래와 같이 문자열의 길이를 계산하는 함수 get_text_length를 정의하자. 이때, docstring은 LLM이 참고하는 정보이므로 함수의 역할을 간단명료하게 작성할 필요가 있다.

def get_text_length(text: str) -> int:
    """주어진 문자열의 길이를 반환하는 함수"""
    return len(text)

 

 LangChain에서 python으로 정의한 함수를 Agent가 사용할 수 있는 LangChain Tool Class로 변환해줘야 한다. 이를 위해 수동으로 작성하는 방법도 있지만 이번에는 tool 데코레이터를 사용해 변환해보자. tool 데코레이터로 함수를 감싸주면, tool의 이름(name), 설명(description) 등이 랭체인 tool class에 채워진다.

from langchain.agents import tool


@tool
def get_text_length(text: str) -> int:
    """주어진 문자열의 길이를 반환하는 함수"""
    text = text.strip("'\n")
    return len(text)


# 확인
print(type(get_text_length))
print(get_text_length.name)
print(get_text_length.description)

 

 아래 결과를 보면, get_text_length는 StructuredTool class이며, name과 description이 parsing되어 클래스 변수로 갖게 된다. 여기서 description에는 함수명, 아규먼트, 반환값 그리고 docstring이 포함되어 있는데, LLM이 이 도구를 사용할지 결정짓는 요소로 작동한다.

<class 'langchain_core.tools.StructuredTool'>
get_text_length
get_text_length(text: str) -> int - 주어진 문자열의 길이를 반환하는 함수

ReAct 프롬프트

 Agent가 문제 해결을 위해 적절한 도구를 선택할 수 있도록 프롬프트를 작성해야 한다. LLM에게 전달한 tool 리스트를 만들고, ReAct 프롬프트 템플릿을 아래와 같이 작성하자. (아래의 프롬프트는 LangChain hub의 hwchase17이 작성한 프롬프트로, ReAct 논문을 구현)

tools = [get_text_length]

template = """
Answer the following questions as best you can. You have access to the following tools:

{tools}

Use the following format:

Question: the input question you must answer
Thought: you should always think about what to do
Action: the action to take, should be one of [{tool_names}]
Action Input: the input to the action
Observation: the result of the action
... (this Thought/Action/Action Input/Observation can repeat N times)
Thought: I now know the final answer
Final Answer: the final answer to the original input question

Begin!

Question: {input}
Thought:
"""

 

 ReAct 알고리즘에서는 위 프롬프트가 LLM에 전달되고, 생성한 LLM의 Thought으로 Tool을 선택하도록 한다.

 

 ReAct 프롬프트는 LLM에게 LLM이 어떻게 생각하고 있는지 어떻게 답을 도출하는지 요청하기 때문에 Chain of Thought 프롬프트의 일종이다.

 

ReAct 프롬프트를 더 자세히 살펴보자.

• {tool}에는 tool 리스트와 description 이 들어감
• {tool_names}에는 tool 리스트에 있는 모든 tool의 이름
• Action Input의 결과로는 get_text_length, argument에는 길이를 알고싶은 대상 text가 될 것이라 예상할 수 있다.
• Observation은 Action의 결과 즉, Tool을 사용한 결과에 대한 관찰이 될 것

위 과정은 N 번 반복될 수 있으며, LLM이 판단하기에 최종 답변이라 생각되면 반복을 멈추고 최종 답변을 반환하도록 한다.

이제, ReAct 템플릿을 활용해서 프롬프트를 만들자.

from langchain.prompts import PromptTemplate
from langchain.tools.render import render_text_description


prompt = PromptTemplate.from_template(template)
prompt = prompt.partial(tools=render_text_description(tools), tool_names=", ".format(t.name for t in tools))

 

여기서, 이미 tool 리스트와 tool 이름은 알고 있는 정보이기 때문에 partial메서드를 사용해 placeholder들({tools}, {tool_names}) 값을 채워준다.

 

tools

• 랭체인 tool 객체로 이루어진 list이므로 string 형식으로 입력받는 LLM 특성 상, tool을 설명하는 문자열로 넣어줘야 한다. 랭체인에는 이를 수행할 수 있도록 render_text_description함수를 제공한다.
• "\n".join([f"{tool.name}: {tool.description}" for tool in tools])과 동일

tool_name

• comma(,)로 구분

따라서, 각 placeholder에 들어가는 문자열은 다음과 같다.

 

tools

    → "get_text_length: get_text_length(text: str) -> int - 주어진 문자열의 길이를 반환하는 함수"

 

tool_names

    →  "get_text_length"

 

사용자 입력인 {input}은 체인이 실행될 때 동적으로 입력받도록 하자.

 

• 최종적으로 LLM에게 전달되는 프롬프트는 다음과 같다.

    Answer the following questions as best you can. You have access to the following tools:    

    get_text_length: get_text_length(text: str) -> int - 주어진 문자열의 길이를 반환하는 함수  

    Use the following format:

    Question: the input question you must answer
    Thought: you should always think about what to do
    Action: the action to take, should be one of [get_text_length]
    Action Input: the input to the action
    Observation: the result of the action
    ... (this Thought/Action/Action Input/Observation can repeat N times)
    Thought: I now know the final answer
    Final Answer: the final answer to the original input question

    Begin!

    Question: 'ReAct'의 길이는?
    Thought:

 

 llm으로 gpt-3.5-turbo를 사용해 해당 프롬프트를 전달하고 응답을 받아보자. 이때, Observation이 나오면 생성을 멈추도록 키워드를 줘서 확인해보자(그렇지 않으면 실제로 Tool을 수행하지 않았는데 할루시네이션이 발생해 계속 생성할 가능성이 있음).

이 부분까지 구현해보자!

llm = ChatOpenAI(temperature=0, model_kwargs={"stop": ["\nObservation"]})

agent = prompt | llm

print(agent.invoke({"input": "'ReAct'의 길이는?"}).content)

I should use the get_text_length function to find the length of the text 'ReAct'.
Action: get_text_length
Action Input: 'ReAct'

Action을 위한 Parsing

Thought, Action, Action Input을 parsing해서 Tool을 실행해보자

 

이제 LLM이 결정한 Action을 수행할 수 있도록 LLM의 출력을 parsing해야 한다. 랭체인에서는 정규표현식을 사용한다. 실제 정규 표현식으로 구현해도 되지만 이 역시 랭체인에서 OutputParser로 구현이 되어 있다. 그 중에서도 ReAct를 위한 OutputParser를 사용해보자. ReActSingleInputOutputParser는 하나의 tool을 제공한 ReAct 스타일의 LLM 응답을 parsing한다.

from langchain.agents.output_parsers import ReActSingleInputOutputParser

agent = prompt | llm | ReActSingleInputOutputParser()

print(agent.invoke({"input": "'ReAct'의 길이는?"}))

tool='get_text_length' tool_input="'ReAct'" 
log="I need to find the length of the text 'ReAct'.\nAction: get_text_length\nAction Input: 'ReAct'"

 

 ReActSingleInputOutputParser클래스의 parse 메서드는 취해야할 action이 있다면 AgentAction(action, tool_input, text)형태로 반환하여 action을 취하려는 tool을 선택하고 실행할 수 있도록 한다. 만약 취할 action이 없다면 AgentFinish 객체를 반환한다.

 

(참고, ReActSingleInputOutputParser의 구현체를 보면 ReAct 프롬프트의 Final Answer 포함여부로 AgentFinish 객체를 반환할지 결정함)

AgentAction 객체와 AgentFinish 객체를 사용해 ReAct Loop의 반복 결정

 

 

 이제, 어떤 Tool을 실행하고 싶은지에 대한 정보를 모두 얻었으므로, LLM이 생각하는 Action을 수행하도록 코드를 구현해보자.

from typing import Union
from langchain.schema import AgentAction, AgentFinish


agent: Union[AgentAction, AgentFinish] = prompt | llm | ReActSingleInputOutputParser()

agent_step = agent.invoke({"input": "'ReAct'의 길이는?"})

# 만약 action을 취해야 한다면, tool과 tool_input을 parsing해 실행
if isinstance(agent_step, AgentAction):
    tool_name = agent_step.tool
    tool_input = agent_step.tool_input
    tool_to_use = find_tool(tools, tool_name)

    observation = tool_to_use.invoke(tool_input)

 

그리고 tool을 찾는 find_tool 함수는 아래와 같이 간단히 구현할 수 있다.

from typing import List
from langchain.tools import Tool


def find_tool(tools: List[Tool], tool_name: str) -> Tool:
    for tool in tools:
        if tool.name == tool_name:
            return tool
    raise ValueError(f"{tool_name}을 가진 Tool을 찾을 수 없습니다.")

 

결과는 다음과 같다.

5

ReAct Loop

 이전까지 Tool이 실행된 후 Output인 observation을 얻었다. 이제 여기서 결과를 반환할지, 아니면 정보가 충분하지 않아 ReAct Loop를 반복해야하는지 결정해야 한다. 이는 ReAct를 실행하며 얻었던 모든 정보를 기록하는 역할을 하는 agent_scratchpad를 사용해 구현한다.

 

 이전까지 결과를 살펴보았을 때는 충분한 결과를 얻었으므로 Agent가 불필요한 step을 밟지 않도록 해보자. 앞서 작성한 프롬프트의 마지막에 {agent_scratchpad}를 추가하자.

template = """
Answer the following questions as best you can. You have access to the following tools:

{tools}

Use the following format:

Question: the input question you must answer
Thought: you should always think about what to do
Action: the action to take, should be one of [{tool_names}]
Action Input: the input to the action
Observation: the result of the action
... (this Thought/Action/Action Input/Observation can repeat N times)
Thought: I now know the final answer
Final Answer: the final answer to the original input question

Begin!

Question: {input}
Thought: {agent_scratchpad}
"""

 

그리고, agent_scratchpad를 리스트 형식으로 정의하고 이를 agent 호출 시 placeholder에 전달해주자. agent_scratchpad에는 ReActSingleInputOutputParser()로 parsing한 tool, tool_input, 그리고 log 값이 인자로 추가된다.

agent_scratchpad는 ReAct 이력을 저장해 Agent가 불필요한 반복을 하지 않도록 하는 역할!

 

 단, 여기서 log를 담고 있는 agent_step은 AgentAction 객체이므로 string으로 type 변환이 필요하다. 이를 수행해주는 format_log_to_str을 사용하자.

format_log_to_str 결과

from langchain.agents.format_scratchpad.log import format_log_to_str


prompt = PromptTemplate.from_template(template)

prompt = prompt.partial(tools=render_text_description(tools), tool_names=", ".join(t.name for t in tools))

llm = ChatOpenAI(
    temperature=0,
    model_kwargs={"stop": ["\nObservation", "Observation"]}
    )

# intermediate_steps 초기화
intermediate_steps = []

# agent chain 정의
agent = prompt | llm | ReActSingleInputOutputParser()

# agent invoke
agent_step = agent.invoke({"input": "'ReAct'의 길이는?",
                           "agent_scratchpad": format_log_to_str(intermediate_steps)
                           })

# 만약 action을 취해야 한다면, tool과 tool_input을 parsing해 실행하고 history를 저장
if isinstance(agent_step, AgentAction):
    tool_name = agent_step.tool
    tool_input = agent_step.tool_input
    tool_to_use = find_tool(tools, tool_name)

    observation = tool_to_use.invoke(tool_input)

    # 과거 실행결과 추가
    intermediate_steps.append((agent_step, observation))

# 과거 실행 이력을 받아 다시 한 번 반복
agent_step = agent.invoke({"input": "'ReAct'의 길이는?",
                           "agent_scratchpad": format_log_to_str(intermediate_steps)})

print(agent_step)

if isinstance(agent_step, AgentFinish):
    print("=== Agent Finish!!===")
    print(agent_step.return_values)

## 1. 첫번째 반복
tool='get_text_length' tool_input="'ReAct'\n" log="I should use the get_text_length function to find the length of the text 'ReAct'.\n    Action: get_text_length\n    Action Input: 'ReAct'\n    "

## 1.1 tool 실행 후 결과를 얻음
observation=5

## 2. 두번째 반복 -> 결과 반환
return_values={'output': '5'} log='I now know the final answer.\nFinal Answer: 5'

=== Agent Finish!!===
{'output': '5'}

 

위에서 보듯 Agent는 여러 번의 LLM 호출이 이루어진다. 위 경우에는 총 2번의 호출이 이뤄졌다.

1. query를 받아 LLM을 호출
2. LLM 응답으로부터 Tool을 parsing해서 Tool 실행 후 결과를 받음
3. 위 이력을 추가(agent_scratchpad)하여 LLM 호출
4. LLM이 판단하기에 최종 답변으로 충분하다고 판단하여 반복 중단
5. 최동 응답 반환

 

 앞서 Agent는 while loop라고 언급했으며, 위 LLM이 판단하기에 최종 답변으로 충분하다고 판단할 때까지 즉, ReActSingleInputOutputParser의 output이 AgentFinish 객체일 때까지 반복한다. 최종적으로 이 글의 초반에서 봤던 zero-shot ReAct Agent example과 같이 작동하는 코드는 다음과 같다.

from typing import List, Union

from dotenv import load_dotenv
from langchain.agents import tool
from langchain.agents.format_scratchpad.log import format_log_to_str
from langchain.agents.output_parsers import ReActSingleInputOutputParser
from langchain.prompts import PromptTemplate
from langchain.schema import AgentAction, AgentFinish
from langchain.tools import Tool
from langchain.tools.render import render_text_description
from langchain_openai import ChatOpenAI

load_dotenv()


@tool
def get_text_length(text: str) -> int:
    """주어진 문자열의 길이를 반환하는 함수"""
    text = text.strip("'\n")
    return len(text)


def find_tool(tools: List[Tool], tool_name: str) -> Tool:
    for tool in tools:
        if tool.name == tool_name:
            return tool
    raise ValueError(f"{tool_name}을 가진 Tool을 찾을 수 없습니다.")


if __name__ == "__main__":
    tools = [get_text_length]

    template = """
    Answer the following questions as best you can. You have access to the following tools:

    {tools}

    Use the following format:

    Question: the input question you must answer
    Thought: you should always think about what to do
    Action: the action to take, should be one of [{tool_names}]
    Action Input: the input to the action
    Observation: the result of the action
    ... (this Thought/Action/Action Input/Observation can repeat N times)
    Thought: I now know the final answer
    Final Answer: the final answer to the original input question

    Begin!

    Question: {input}
    Thought: {agent_scratchpad}
    """

    prompt = PromptTemplate.from_template(template)
    prompt = prompt.partial(
        tools=render_text_description(tools),
        tool_names=", ".join(t.name for t in tools),
    )

    llm = ChatOpenAI(
        temperature=0,
        model_kwargs={"stop": ["\nObservation", "Observation"]},
    )

    intermediate_steps = []

    agent = prompt | llm | ReActSingleInputOutputParser()

    # Invoke
    agent_step = None

    while not isinstance(agent_step, AgentFinish):
        agent_step = agent.invoke(
            {
                "input": "'ReAct'의 길이는?",
                "agent_scratchpad": format_log_to_str(intermediate_steps),
            }
        )

        print(agent_step)

        if isinstance(agent_step, AgentAction):
            tool_name = agent_step.tool
            tool_input = agent_step.tool_input
            tool_to_use = find_tool(tools, tool_name)
            observation = tool_to_use.invoke(tool_input)

            print(f"{observation=}")

            intermediate_steps.append((agent_step, str(observation)))

        if isinstance(agent_step, AgentFinish):
            print("=== Agent Finish!!===")
            print(agent_step.return_values)

tool='get_text_length' tool_input="'ReAct'\n" log="I should use the get_text_length function to find the length of the text 'ReAct'.\n    Action: get_text_length\n    Action Input: 'ReAct'\n    "

observation=5

return_values={'output': '5'} log='I now know the final answer is 5. \n\nFinal Answer: 5'

=== Agent Finish!!===
{'output': '5'}