[GenAI StudyJam] Chain of Thought React (CoT)

본 글은 Google Cloud Skills Boost의 Beginner:Introduction to Generative AI Learning Path > Prompt Design in Vertex AI: Challenge Lab 에서 배운 내용을 정리한 글이며, 본 글에 첨부된 코드의 원본은  GoogleCloudPlatform/generative-ai 에서 확인하실 수 있습니다.

 

Chain of Thougt React

Tools

ReAct agent

• 상호 대화 시스템을 만들기 위해 디자인된 언어 모델의 한 종류로, 텍스트 생성, 번역, 창의적인 콘텐츠 생성과 정보를 얻고 주기 위한 질의응답 등이 가능하다.

LangChain

• 개발하고 배포하기 위해 다양한 사전훈련 언어모델, 간단한 API, 다양한 task 등 다양한 feature들을 제공하여 ReAct agent를 좀 더 쉽게 만들 수 있게 하는 라이브러리이다.

Vertex AI 

• 머신러닝 모델을 더 쉽게 만들고 훈련하고 배포하기 위한 머신러닝 플랫폼. 사전 훈련된 언어 모델, 간단한 API 등 다양한 task를 지원한다.

!pip install --user langchain==0.0.310 \
                    google-cloud-aiplatform==1.35.0 \
                    prettyprinter==0.18.0 \
                    wikipedia==1.4.0 \
                    chromadb==0.3.26 \
                    tiktoken==0.5.1 \
                    tabulate==0.9.0 \
                    sqlalchemy-bigquery==1.8.0 \
                    google-cloud-bigquery==3.11.4

# 필요한 라이브러리들을 import 해준다.
import vertexai
import os
import IPython
from langchain.llms import VertexAI
from IPython.display import display, Markdown


PROJECT_ID = "your-project-here"  # @param {type:"string"} # GCP의 프로젝트 ID가 여기 들어가야 한다.
LOCATION = "us-central1"  # @param {type:"string"} # GCP Region을 설정해줘야 한다.

vertexai.init(project=PROJECT_ID, location=LOCATION) # Vertexai 초기화

Introduction

• Chain of Thought은 LLM의 추론 능력을 강화하기 위한 것으로, 특히 추론에 여러 단계의 추론과정이 필요한 task (주로 복잡한 문제)에 사용된다.
• 모델이 바로 최종 답변을 내놓게 하는 기존의 프롬프팅과 달리 Chain of Thought Prompting은 LLM이 최종 답변을 내놓기 전에 문제에 대한 중간 추론 과정을 생성하게 하여 복잡한 문제를 좀 더 간단한 중간 단계로 쪼개서 직관적으로 복잡한 과정의 문제를 푸는 사람의 생각 과정을 흉내낼 수 있게 한다. 

  

CoT의 사용 사례

CoT의 구현

# 사용할 모델은 Google PaLM2 Bison 모델로, 언어 이해, 생성, 대화에서 뛰어난 역량을 보여주는 LLM이라고 한다.
MODEL_NAME = "text-bison@001"
llm = VertexAI(model_name=MODEL_NAME, max_output_tokens=1000)

Chain of Thought 구현

• 보다시피 답이 어떻게 나오게 되었는지 추론하는 과정을 예시로 함께 프롬프트에 제공해줌으로써 모델이 답안과 비슷한 추론 과정을 답과 함께 낼 수 있도록 한다.

question = """Q: Roger has 5 tennis balls. He buys 2 more cans of tennis balls.
Each can has 3 tennis balls. How many tennis balls does he have now?
A: The answer is 11.
Q: The cafeteria had 23 apples.
If they used 20 to make lunch and bought 6 more, how many apples do they have?
A:"""

llm.predict(question)

Zero Shot CoT prompting

• LLM이 질문에 대한 더 정확한 답변을 만들어내도록하는 기술로, 'Let's think step by step'과 같은 말을 질문 후 답변 예시에 덧붙여 주는 간단한 형식으로 구현할 수 있다. 

  

question = """Q: Roger has 5 tennis balls. He buys 2 more cans of tennis balls. Each can has 3 tennis balls. 
How many tennis balls does he have now? 
A: The answer is 11. 
Q: The cafeteria had 23 apples. If they used 20 to make lunch and bought 6 more, 
how many apples do they have? 
A: Let's think step by step.""" 

llm.predict(question)

Self Consistency

• 같은 입력에 대해서 LLM이  CoT를 통한 다양한 후보답안을 내게 함으로써 프롬프트를 강화한다.

from operator import itemgetter
from langchain.prompts import PromptTemplate
from langchain.schema import StrOutputParser
from langchain.schema.runnable import RunnablePassthrough

 

여기서부터는 예시로 주는 문장들을 따로 작성해 template을 만들어 모델에 질문한다.

question = """The cafeteria had 23 apples.
If they used 20 to make lunch and bought 6 more, how many apples do they have?"""

context = """Answer questions showing the full math and reasoning.
Follow the pattern in the example.
"""

one_shot_exemplar = """Example Q: Roger has 5 tennis balls. He buys 2 more cans of tennis balls.
Each can has 3 tennis balls. How many tennis balls does he have now?
A: Roger started with 5 balls. 2 cans of 3 tennis balls
each is 6 tennis balls. 5 + 6 = 11.
The answer is 11.

Q: """

 

여기서부터 LangChain Expression Language를 알 필요가 있는데, unix에서 파이프 연산을 하는 것과 비슷하게 생각하면 된다.

planner = ( 프롬프트 | 모델 | 출력 파서 | 전달할 출력 데이터 )
answer = (프롬프트 + 전달 받은 답변 | 모델 | 출력 파서 )

 

아래의 코드는 위의 코드 방식으로 전개된다고 보면 된다.

여러 개의 답변 후보들을 받기 위해 temperature를 다양하게 조정한 모델 3 개에게 기존 프롬프트에 기존의 model이 내놓은 답변을 예시로 추가해 전달한다.

참고로 temperature는 대규모 언어모델의 하이퍼 파라미터로, 숫자가 높아질수록 모델이 형식적인 답변에서 벗어나 창의적인 답변을 내놓을 가능성이 높아지는 것을 의미한다.

planner = (
    PromptTemplate.from_template(context + one_shot_exemplar + " {input}") # from_template() 메소드로 PromptTemplate 객체 생성
    | VertexAI()
    | StrOutputParser()
    | {"base_response": RunnablePassthrough()} # base_response을 다음 입력으로 넘기겠다는 의미
)

# temperature = 0
answer_1 = (
    PromptTemplate.from_template("{base_response} A: 33") # 기존의 
    | VertexAI(temperature=0, max_output_tokens=400)
    | StrOutputParser()
)

# temperature = 0.1
answer_2 = (
    PromptTemplate.from_template("{base_response} A:")
    | VertexAI(temperature=0.1, max_output_tokens=400)
    | StrOutputParser() # string 타입 output 생성
)

# temperature = 0.7
answer_3 = (
    PromptTemplate.from_template("{base_response} A:")
    | VertexAI(temperature=0.7, max_output_tokens=400)
    | StrOutputParser()
)

# 모든 답변을 종합해 최종 답변으로 내도록 한다.
final_responder = (
    PromptTemplate.from_template(
        "Output all the final results in this markdown format: Result 1: {results_1} \n Result 2:{results_2} \n Result 3: {results_3}"
    )
    | VertexAI(max_output_tokens=1024)
    | StrOutputParser()
)

# 체인을 구성해준다.
chain = (
    planner
    | {
        "results_1": answer_1,
        "results_2": answer_2,
        "results_3": answer_3,
        "original_response": itemgetter("base_response"), # map에서 base_response 추출
    }
    | final_responder
)

# 체인 실행 후 모델이 낸 답을 마크다운 형식으로 보여준다.
answers = chain.invoke({"input": question})
display(Markdown(answers))

ReAct (Reasoning & Acting)

Introduction

• ReAct는 CoT와 외부 tool을 함께 사용하는 것으로, 복잡한 문제를 추론하는 과정에서 외부의 tool과 상호작용하는 시스템이다. 대규모 언어 모델이나 대규모 언어모델 기반 챗봇이 확장을 통해 외부 시스템으로 행동하고 추론하기를 원할 때 사용한다.

ReAct의 구현

from langchain.tools import StructuredTool
from langchain.agents import AgentType, initialize_agent, load_tools
from langchain.llms import VertexAI
from langchain.tools import WikipediaQueryRun
from langchain.utilities import WikipediaAPIWrapper
import wikipedia
import vertexai

 

예를 들어, 언어모델이 오늘 날짜를 모른다고 하면, 따로 날짜를 원하는 포맷으로 추출하는 외부 함수를 만들 수 있다.

# 오늘 날짜를 YYYY-MM-DD의 포맷으로 리턴한다.
def get_current_date():

	from datetime import datetime

    todays_date = datetime.today().strftime("%Y-%m-%d")

    return todays_date

 

이 함수를 ReAct Agent에 tool로 넘겨 사용할 수 있게 한다.

# 주어진 함수로부터 tool 생성
t_get_current_date = StructuredTool.from_function(get_current_date)

# 사용할 tool 지정
tools = [
    t_get_current_date,
]

# 주어진 tool과 LLM을 사용하기 위한 agent 실행자 로드
agent = initialize_agent(
    tools,
    llm,
    agent=AgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION, # acting 전 resoning 스텝을 거치는 zero-shot agent
    verbose=True,
)

# 실행
agent.run("What's today's date?")

action 전에 내가 정의한 함수를 사용하는 것과 thought에 모델의 사고 과정을 확인할 수 있다.

Wikipedia-ReAct

• 모델이 질문에 대해 응답 할 때 wikipedia를 참고하여 응답하게 할 수 있다.

# VertexAI LLM을 모델로 사용 - 모델의 창의적인 답이 아니라 문서에 기반한 답을 원하므로, temperature = 0으로 설정
llm = VertexAI(temperature=0)

# Wikipedia API 검색 툴
_ = WikipediaQueryRun(api_wrapper=WikipediaAPIWrapper())

# Wikipedia API를 tool로 넘겨줌 
tools = load_tools(["wikipedia"], llm=llm)

# 정의했던 날짜 함수도 함께 tool에 넘겨주기 
tools.append(t_get_current_date)

# Agent 정의
agent = initialize_agent(
    tools,
    llm,
    agent=AgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True,
)

agent.run(
    "Fetch today's date, and tell me which famous person was born or who died on the same day as today?"
)

 

tool로 줬던 함수에서 오늘 날짜를 받은 후 위키피디아에서 해당 날짜에 죽거나 태어난 사람들을 찾는 모습을 확인할 수 있다.

BigQuery-ReAct

• 특정 데이터셋에서 query를 이용해 데이터를 불러와 가공하는 tool 함수를 정의하고, 이를 이용해 데이터를 불러와 분석하는 reAct Agent를 만들 수 있다.

import re
from typing import Sequence, List, Tuple, Optional, Any
from langchain.agents.agent import Agent, AgentOutputParser
from langchain.prompts.prompt import PromptTemplate
from langchain.prompts.base import BasePromptTemplate
from langchain.tools.base import BaseTool
from langchain.agents import Tool, initialize_agent, AgentExecutor
from langchain.llms import VertexAI
from langchain.agents.react.output_parser import ReActOutputParser
import pandas as pd
from google.cloud import bigquery
from langchain.output_parsers import CommaSeparatedListOutputParser
from langchain.tools.python.tool import PythonREPLTool

# Project-ID에는 GCP 프로젝트 아이디를 기입한다.
bq = bigquery.Client(project=PROJECT_ID)
# 마찬가지로, 모델이 답을 만들어내는 것을 금지하기 위해 temperature = 0인 LLM을 사용한다.
llm = VertexAI(temperature=0, max_tokens=1024)

 

이번에 전해줄 tool은 두 가지이다. 아래와 같이 Bigquery문을 이용해 데이터를 뽑아오는 함수와 앞에서 만들었던 오늘 날짜를 리턴하는 함수이다. 이 두가지 tool을 이용해 특정 날짜와 관련된 사람을 위키피디아에서 뽑아오는 작업을 할 수 있는 ReAct Agent를 만들자.

# BigQuery를 이용해 id로 댓글을 찾아 추출하여 데이터프레임으로 변환하는 함수
def get_comment_by_id(id: str) -> str:
    QUERY = "SELECT text FROM bigquery-public-data.hacker_news.full WHERE ID = {id} LIMIT 1".format(
        id=id
    )
    df = bq.query(QUERY).to_dataframe()

    return df

# BigQuery를 이용해 user로 댓글을 찾아 추출하여 데이터프레임으로 변환하는 함수
def get_comment_by_user(user):
    QUERY = "SELECT text FROM bigquery-public-data.hacker_news.full WHERE `BY` = {user} LIMIT 10".format(
        user=user
    )
    df = bq.query(QUERY).to_dataframe()

    return df

# LLM에게 주어진 댓글에 친절한 답글을 생성하도록 하는 함수
def generate_response_for_comment(comment):
    question = """Create a 1 sentence friendly response to the following comment: {comment}""".format(
        comment=comment
    )
    llm1 = VertexAI(temperature=0.3, max_output_tokens=150)
    response = llm1.predict(question)

    return response

# LLM을 이용해 댓글에 대한 감성 분류를 하는 함수
def generate_sentiment_for_comment(comment):
    question = """What is the sentiment of the comment (Negative, Positive, Neutral): {comment}""".format(
        comment=comment
    )
    llm1 = VertexAI(temperature=0.3, max_output_tokens=150)
    response = llm1.predict(question)

    return response

# LLM을 이용해 댓글의 카테고리 분류를 하는 함수
def generate_category_for_comment(comment):
    question = """Put the comment into one of these categories (Technology, Politics, Products, News): {comment}""".format(
        comment=comment
    )
    llm1 = VertexAI(temperature=0.3, max_output_tokens=150)
    response = llm1.predict(question)

    return response

 

 

아래와 같이 작성했던 모든 함수들을 name과 description과 함께 tool로 정의해준다.

tools = [
    Tool(
        name="GetCommentsById",
        func=get_comment_by_id,
        description="Get a pandas dataframe of comment by id.",
    ),
    Tool(
        name="GetCommentsByUser",
        func=get_comment_by_user,
        description="Get a pandas dataframe of comments by user.",
    ),
    Tool(
        name="GenerateCommentResponse",
        func=generate_response_for_comment,
        description="Get an AI response for the user comment.",
    ),
    Tool(
        name="GenerateCommentSentiment",
        func=generate_sentiment_for_comment,
        description="Get an AI sentiment for the user comment.",
    ),
    Tool(
        name="GenerateCategorySentiment",
        func=generate_category_for_comment,
        description="Get an AI category for the user comment.",
    ),
    PythonREPLTool(),
]

 

아래와 같이 prompt template을 만든다.

# 다음과 같이 Agent의 CoT 과정을 예시로 함께 넘겨준다.
EXAMPLES = [
    """Question: Write a response to the following Comment 1234 ?
Thought: I need to get comment 1234 using GetCommentsById.
Action: GetCommentsById[1234]
Observation: "Comment Text"
Thought: I need to generate a response to the comment.
Action: GenerateCommentResponse["Comment Text"]
Observation: LLM Generated response
Thought: So the answer is "LLM Generated response".
Action: Finish["LLM Generated response"],
Question: Write a response to all the comments by user xx234 ?
Thought: I need to get all the comments by xx234 using GetCommentsByUser.
Action: GetCommentsByUser['xx234']
Observation: "Comment Text"
Thought: I need to generate a response to each comment.
Action: GenerateCommentResponse["Comment Text 1"]
Observation: "LLM Generated response 1"
Thought: I need to generate a response to each comment.
Action: GenerateCommentResponse["Comment Text 2"]
Observation: "LLM Generated response 2"
Thought: I need to generate a response to each comment.
Action: GenerateCommentResponse["Comment Text 3"]
Observation: "LLM Generated response 3"
Thought: I Generated responses for all the comments.
Action: Finish["Done"],
Question: Sentiment for all the comments by user xx234 ?
Thought: I need to get all the comments by xx234 using GetCommentsByUser.
Action: GetCommentsByUser['xx234']
Observation: "Comment Text"
Thought: I need to determine sentiment of each comment.
Action: GenerateCommentSentiment["Comment Text 1"]
Observation: "LLM Generated Sentiment 1"
Thought: I need to determine sentiment of each comment.
Action: GenerateCommentSentiment["Comment Text 2"]
Observation: "LLM Generated Sentiment 2"
Thought: I need to generate a response to each comment.
Action: GenerateCommentSentiment["Comment Text 3"]
Observation: "LLM Generated Sentiment 3"
Thought: I determined sentiment for all the comments.
Action: Finish["Done"],
Question: Category for all the comments by user xx234 ?
Thought: I need to get all the comments by xx234 using GetCommentsByUser.
Action: GetCommentsByUser['xx234']
Observation: "Comment Text"
Thought: I need to determine the category of each comment.
Action: GenerateCategorySentiment["Comment Text 1"]
Observation: "LLM Generated Category 1"
Thought: I need to determine category of each comment.
Action: GenerateCategorySentiment["Comment Text 2"]
Observation: "LLM Generated Category 2"
Thought: I need to generate a category to each comment.
Action: GenerateCategorySentiment["Comment Text 3"]
Observation: "LLM Generated Category 3"
Thought: I determined Category for all the comments.
Action: Finish["Done"]
"""
]

SUFFIX = """\nIn each action, you cannot use the nested functions, such as GenerateCommentResponse[GetCommentsByUser["A"], GetCommentsById["B"]].
Instead, you should parse into 3 actions - GetCommentsById['A'], GetCommentsByUser['B'], and GenerateCommentResponse("Comment").

Let's start.

Question: {input}
{agent_scratchpad} """

output_parser = CommaSeparatedListOutputParser()

format_instructions = output_parser.get_format_instructions()

TEST_PROMPT = PromptTemplate.from_examples(
    examples=EXAMPLES,
    suffix=SUFFIX,
    input_variables=["input", "agent_scratchpad"], # agent_scratchpad는 모델의 사고과정을 추적하는 데에 사용된다.
)

 

아래처럼 입맛대로 Langchain 함수들을 커스텀할 수 있다.

class ReActTestAgent(Agent):
	# 커스텀한 output parser를 agent에 전달, AgentOUtputParser를 결과값으로 리턴
    @classmethod
    def _get_default_output_parser(cls, **kwargs: Any) -> AgentOutputParser:
        return ReActOutputParser()
	
    # agent에게 넘길 prompt template을 생성하는 메소드 (앞서 정의한 TEST_PROMPT를 기반으로 맞춤형 프롬프트 제작)
    @classmethod
    def create_prompt(cls, tools: Sequence[BaseTool]) -> BasePromptTemplate:
        return TEST_PROMPT
	
    # 6개의 tool이 제공이 되었는지 확인하고, 미리 정의된 set에 맞는지 확인하여 agent가 해당 tool들을 넘겨받았는지 확인
    @classmethod
    def _validate_tools(cls, tools: Sequence[BaseTool]) -> None:
        if len(tools) != 6:
            raise ValueError("The number of tools is invalid.")
        tool_names = {tool.name for tool in tools}
        if tool_names != {
            "GetCommentsById",
            "GetCommentsByUser",
            "GenerateCommentResponse",
            "GenerateCommentSentiment",
            "GenerateCategorySentiment",
            "Python_REPL",
        }:
            raise ValueError("The name of tools is invalid.")
	
    ## Agent property
    # agent type을 string 타입으로 반환
    @property
    def _agent_type(self) -> str:
        return "react-test"
	
    # 결과값에서 최종 답안을 가리키기 위한 문자열
    @property
    def finish_tool_name(self) -> str:
        return "Final Answer: "
	
    # observation: Bigquery에서 관찰한 사실 기재를 명시하는 문자열
    @property
    def observation_prefix(self) -> str:
        return f"Observation: "
	
    # LLM의 사고과정을 명시하는 prefix 
    @property
    def llm_prefix(self) -> str:
        return f"Thought: "

 

agent를 정의하고 실행하도록 구성한다.

llm = VertexAI(
    temperature=0,
)

agent = ReActTestAgent.from_llm_and_tools(llm, tools, verbose=True) # 커스텀한 agent 클래스 사용

agent_executor = AgentExecutor.from_agent_and_tools(
    agent=agent, tools=tools, verbose=True
)
agent_executor.handle_parsing_errors = True

 

사용할 때는 다음과 같은 방식으로 사용할 수 있다. 이를 통해서 따로 Query로 데이터를 가져오고 할 필요 없이 바로 감성분류까지 한 방에 되니까 정말 편리한 툴이 아닐 수 없다.

input = "Get the category for comment 8885404"
agent_executor.run(input)

똑같은 댓글만 반복해서 꺼내다가 could not parse LLM output이 나오더니, time limit을 넘어서서 아쉽게도 final ouput까지는 가지 못했다.

 

특정 사용자가 적은 댓글에 대한 감성 분류를 해달라고 하거나, 댓글에 대한 답글을 적어달라고 명령 내릴 수도 있다.

input = "Get the sentiment for all the to comments written by chris"
agent_executor.run(input)

input = "Get the category for all the to comments written by chris"
agent_executor.run(input)

input = "Get the response for all the to comments written by chris."
agent_executor.run(input)

 

LangChain 배워두면 LLM 이용한 프로젝트 할 때 정말 쓸만하겠다.