[LangChain] Agent와 Tool 사용법

LangChain에서 Agent는 주어진 쿼리에 대해 스스로 생각하여 적절한 행동을 선택하는 대리인이라고 보면 된다.

 

Agent는 자신의 목표를 달성할 때까지 자신에게 주어진 Tool들을 기반으로 적절한 Action을 수행한다.

 

한번 유명인의 나이를 가지고 수학 연산을 해보자.

 

Tool

필요한 tool은 두 가지이다.

1. 유명인에 대한 검색 : 최신 정보를 반영하기 위해 검색 api를 사용한다.
2. 수학 연산 : 복잡한 수학 연산을 위해

tool을 정의해보자.

 

tool을 정의하는 방법에도 여러가지가 있다.

 

1. Tool 데이터 클래스 사용

Tool.from_function(
    func=llm_math_chain.run,
    name="Calculator",
    description="useful for when you need to answer questions about math",
    args_schema=CalculatorInput,
    # coroutine= ... <- you can specify an async method if desired as well
)

 

2. BaseTool 하위 클래스

class CustomSearchTool(BaseTool):
    name = "custom_search"
    description = "useful for when you need to answer questions about current events"

    def _run(
        self, query: str, run_manager: Optional[CallbackManagerForToolRun] = None
    ) -> str:
        """Use the tool."""
        return search.run(query)

    async def _arun(
        self, query: str, run_manager: Optional[AsyncCallbackManagerForToolRun] = None
    ) -> str:
        """Use the tool asynchronously."""
        raise NotImplementedError("custom_search does not support async")

 

3. tool 데코레이터 사용

@tool
def search_api(query: str) -> str:
    """Searches the API for the query."""
    return f"Results for query {query}"

 

 

이 중 3번 tool 데코레이터를 사용해서 우리가 사용할 tool을 직접 정의해보자.

 

참고로 tool 데코레이터를 사용할 때는 반드시 docstring을 사용해서 이 tool에 대한 설명을 적어줘야 한다.

 

구글 검색은 serp api를 사용할 예정이고 수학 계산은 LLMMathChain을 사용할 예정이다.

from langchain.tools import tool
from langchain.utilities import SerpAPIWrapper
from langchain.chains import LLMMathChain
chain = LLMMathChain(llm=chat_model, verbose=True)

# serp api
# google search를 위해서 google-search-results 패키지를 설치해야 한다.
# SERPAPI_API_KEY에 serpapi key 값을 환경 변수로 등록해야 한다.
@tool("search")
def search_api(query : str) -> str:
    """Searchs the api for the query""" # tool decorator를 사용하면 docstring으로 이에 대한 설명을 적는다.
    search = SerpAPIWrapper()
    result = search.run(query)
    return result

@tool("math")
def math(query : str) -> str:
    """useful for when you need to answer questions about math"""
    llm_math_chain = LLMMathChain(llm=chat_model, verbose=True)
    return llm_math_chain.run(query)

tools = [search_api, math]

참고로 우리가 openai chat model을 사용할 때 api key를 환경 변수에 등록한 것 처럼 serp api를 사용하기 위해서는 역시 똑같이 api key를 환경 변수로 등록해야 한다. 

 

SERPAPI_API_KEY를 키로 하고 홈페이지에서 받은 api key를 값으로 환경 변수를 등록하자.

https://serpapi.com/

SerpApi: Google Search API

 

(OPENAI_API_KEY 등록 과정 참고)

https://growth-coder.tistory.com/253

[LangChain] LangChain 개념 및 사용법

 

Prompt

prompt를 정의한다.

# prompt 정의
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder

prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You are very powerful assistant, but bad at calculating lengths of words.",
        ),
        ("user", "{input}"),
        MessagesPlaceholder(variable_name="agent_scratchpad"), # 중간 과정 전달
    ]
)

agent_scratchpad에는 중간 과정을 리스트로 전달해주면 된다.

 

보통 아래와 같이 AgentAction과 그 결과가 리스트 형태로 저장되어 있다.

 

List[Tuple[AgentAction, Any]]

 

tool을 사용할 수 있게 openai function으로 변경해서 chat model에 바인딩한다.

# tool을 사용할 수 있게 openai function으로 변경
from langchain.tools.render import format_tool_to_openai_function
openai_functions = [format_tool_to_openai_function(t) for t in tools]
print(openai_functions[0])
chat_model_with_tools = chat_model.bind(functions = [format_tool_to_openai_function(t) for t in tools])

 

Agent

이제 agent를 생성한다.

# agent 생성
from langchain.schema.runnable.passthrough import RunnablePassthrough
from langchain.schema.runnable import RunnableLambda
from langchain.agents.output_parsers import OpenAIFunctionsAgentOutputParser
from langchain.agents.format_scratchpad import format_to_openai_functions
from operator import itemgetter
agent = (
    RunnablePassthrough.assign(
        agent_scratchpad=RunnableLambda(
            itemgetter('intermediate_steps')
        )
        | format_to_openai_functions
    )
    | prompt
    | chat_model_with_tools
    | OpenAIFunctionsAgentOutputParser()
)

간단한 query를 invoke 해보자.

# query 
res = agent.invoke({"input": "Who is Leo DiCaprio's girlfriend? What is her current age raised to the 0.43 power?", "intermediate_steps": []})
print(res)

결과 
tool='search' tool_input={'query': "Leo DiCaprio's girlfriend"} log='\nInvoking: `search` with `{\'query\': "Leo DiCaprio\'s girlfriend"}`\n\n\n' message_log=[AIMessage(content='', additional_kwargs={'function_call': {'name': 'search', 'arguments': '{\n "query": "Leo DiCaprio\'s girlfriend"\n}'}})]

디카프리오의 여자친구를 검색하기 위해서 가장 처음 search tool을 선택한 모습을 볼 수 있다.

 

Runtime

 

이제 원하는 결과를 얻기 위해 반복문을 통해 계속 쿼리를 날려보자.

# 반복문으로 원하는 결과를 찾을 때까지 반복
from langchain.schema.agent import AgentFinish

user_input = "Who is Leo DiCaprio's girlfriend? What is her current age raised to the 0.43 power?"
intermediate_steps = []
while True:
    output = agent.invoke(
        {
            "input": user_input,
            "intermediate_steps": intermediate_steps,
        }
    )
    if isinstance(output, AgentFinish): # 끝났다면
        final_result = output.return_values["output"]
        break
    else:
        # 어떤 tool을 선택했는지 출력
        print(f"TOOL NAME: {output.tool}")
        print(f"TOOL INPUT: {output.tool_input}")
        # output.tool은 선택한 tool의 이름이다.
        # key가 tool 이름, value가 tool인 dictionary를 정의해서 해당하는 tool을 가져온다.
        tool = {
            "search": search_api,
            "math" : math
            }[output.tool]
        # 선택한 tool 실행
        observation = tool.run(output.tool_input)
        # 중간 과정 저장 
        intermediate_steps.append((output, observation))
print(final_result)
print(intermediate_steps)

검색을 위한 search, 계산을 위한 math tool이 사용되었다는 것을 알 수 있다.

최종 결과는 아래와 같다.

Leonardo DiCaprio's girlfriend is Vittoria Ceretti. Her current age raised to the 0.43 power is approximately 4.059.

 

그리고 intermediate_steps를 출력해서 중간 과정이 잘 저장되었는지 확인해보자.

[(AgentActionMessageLog(tool='search', tool_input={'query': 'Leo DiCaprio girlfriend'}, log="\nInvoking: `search` with `{'query': 'Leo DiCaprio girlfriend'}`\n\n\n", message_log=[AIMessage(content='', additional_kwargs={'function_call': {'name': 'search', 'arguments': '{\n"query": "Leo DiCaprio girlfriend"\n}'}})]), 'In August, Leonardo DiCaprio was seen with Italian model Vittoria Ceretti, despite rumors of a potential relationship with model Gigi Hadid over the past year. New rumors started about Ceretti this summer, when she was seen on a yacht with DiCaprio and other models Meghan Roche and Imaan Hammam.'),
 (AgentActionMessageLog(tool='math', tool_input={'query': '26^0.43'}, log="\nInvoking: `math` with `{'query': '26^0.43'}`\n\n\n", message_log=[AIMessage(content='', additional_kwargs={'function_call': {'name': 'math', 'arguments': '{\n"query": "26^0.43"\n}'}})]), 'Answer: 4.059182145592686')]

search와 math tool 사용했던 흔적을 확인할 수 있다.

 

우리는 우리가 직접 반복문을 통해 원하는 과정을 찾을 때까지 반복했고 직접 중간 과정을 출력하는 코드를 작성해서 Runtime을 구현했다.

 

이러한 과정을 매번 구현하기에는 번거로운 일이다. 이럴 때 agent executor를 사용하면 쉽게 이러한 과정을 구현할 수 있다.

 

Agent executor

# agent executor 사용
from langchain.agents import AgentExecutor
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
res = agent_executor.invoke({"input": "Who is Leo DiCaprio's girlfriend? What is her current age raised to the 0.43 power?"})
print(res)

 

이렇게 agent executor를 활용하면 쉽게 구현할 수 있다.

 

참고

https://python.langchain.com/docs/modules/agents/

Agents | 🦜️🔗 Langchain