목차
- 작성 개요
- 도구 및 작업 실행 에이전트 정의
- 상태 정의
- 계획 단계 및 재계획 단계
- 노드 정의
- 그래프 생성 및 실행
- 함께 읽어보면 좋은 논문과 후기
1. 작성 개요
다음 게시물은 LangGraph를 이용한 UseCase 중 하나인 Plan and Execute에 대한 정리글이다. 우선 Plan and Execute는 두가지 특징을 갖는다.
- 장기 계획 수립: 명시적으로 장기 계획을 수행하여 안정적인 진행이 가능하도록 함
- 단계 별 실행 및 재계획: 단계적으로 계획을 세워 실행하고, 계획이 유효한지 검토한 후 수정할 수 있다.
- 효율적인 모델 사용: 계획 단계에서 비싼 모델을 사용하고, 실행단계에서 저렴한 모델을 사용하여 효율성을 확보할 수 있음
따라서 논문과 정확히 일치 하지 않지만, 다음 구조를 어떻게 구현하는지 코드로 정리할 예정이다.
다음 게시물은 테디노트님의 강의를 듣고 정리되었다. 게시물의 내용이 부족하다고 여겨진다면 꼭 유료 강의를 구매하여 듣는 것을 권장한다.
2. 도구 및 작업 실행 에이전트 정의
우선 도구는 웹검색 도구 1개만 사용될 것이다.
from langchain_teddynote.tools import TavilySearch
# Tavily 검색 도구 초기화
tools = [TavilySearch(max_results=3)]또 도구를 바탕으로 작업을 실행할 에이전트를 정의할 수 있는데, 여기서 주목할 점은 langgraph에 포함되어 있는 create_react_agent를 사용한다는 점이다.
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent
from langchain_core.prompts import ChatPromptTemplate
# 프롬프트 정의
prompt = ChatPromptTemplate.from_messages(
[
(
"system",
"You are a helpful assistant. Answer in Korean.",
),
("human", "{messages}"),
]
)
# LLM 정의
llm = ChatOpenAI(model=MODEL_NAME, temperature=0)
# ReAct 에이전트 생성
agent_executor = create_react_agent(llm, tools, state_modifier=prompt)
3. 상태 정의
langgraph인 만큼 당연히 상태를 정의해야 한다.
import operator
from typing import Annotated, List, Tuple
from typing_extensions import TypedDict
# 상태 정의
class PlanExecute(TypedDict):
input: Annotated[str, "User's input"]
plan: Annotated[List[str], "Current plan"]
past_steps: Annotated[List[Tuple], operator.add]
response: Annotated[str, "Final response"]상태에서 주목할만한 점은 plan과 past_steps다. plan은 현재 계획인데, 계획이 하나의 항목으로만 있지는 않기 때문에 List[str]이다. 리스트 안에 여러가지의 문자열로 된 계획을 넣을 뜻이라는 것.
past_step은 리스트 안에 튜플이 들어있는데, 이전에 실행한 계획과 실행 결과가 담겨 있는 것이다. 단순히 계획들만 담겨 있는 것이 아니라 계획들에 대한 실행 결과도 같이 담겨 있어야 하기 때문에, 튜플로 이를 묶어야 한다.
4. 계획 단계 및 재계획 단계
우선 기본적인 계획을 생성해주는 단계가 필요하다.
from pydantic import BaseModel, Field
from typing import List
# Plan 모델 정의
class Plan(BaseModel):
"""Sorted steps to execute the plan"""
steps: Annotated[List[str], "Different steps to follow, should be in sorted order"]
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
# 계획 수립을 위한 프롬프트 템플릿 생성
planner_prompt = ChatPromptTemplate.from_messages(
[
(
"system",
"""For the given objective, come up with a simple step by step plan. \
This plan should involve individual tasks, that if executed correctly will yield the correct answer. Do not add any superfluous steps. \
The result of the final step should be the final answer. Make sure that each step has all the information needed - do not skip steps.
Answer in Korean.""",
),
("placeholder", "{messages}"),
]
)
planner = planner_prompt | ChatOpenAI(
model=MODEL_NAME, temperature=0
).with_structured_output(Plan)
# Planner 실행
planner.invoke(
{
"messages": [
(
"user",
"LangGraph 의 핵심 장단점과 LangGraph 를 사용하는 이유는 무엇인가?",
)
]
}
)
위처럼 데이터 모델 클래스를 작성해 원하는 형식대로 유도된 답변을 할 수 있게 한다면, 질문에 맞는 대답을 반환하는데 필요한 정보를 찾는 계획을 세울 수 있게 된다.
또 계획을 통한 실행이 적절히 이루어지지 않을 수 있기 때문에, 그런 경우에 계획을 새로 세워야 한다. 이런 경우를 위해 재계획 단계가 필요하다.
from typing import Union
class Response(BaseModel):
"""Response to user."""
# 사용자 응답
response: str
class Act(BaseModel):
"""Action to perform."""
# 수행할 작업: "Response", "Plan". 사용자에게 응답할 경우 Response 사용, 추가 도구 사용이 필요할 경우 Plan 사용
action: Union[Response, Plan] = Field(
description="Action to perform. If you want to respond to user, use Response. "
"If you need to further use tools to get the answer, use Plan."
)
# 계획을 재수립하기 위한 프롬프트 정의
replanner_prompt = ChatPromptTemplate.from_template(
"""For the given objective, come up with a simple step by step plan. \
This plan should involve individual tasks, that if executed correctly will yield the correct answer. Do not add any superfluous steps. \
The result of the final step should be the final answer. Make sure that each step has all the information needed - do not skip steps.
Your objective was this:
{input}
Your original plan was this:
{plan}
You have currently done the follow steps:
{past_steps}
Update your plan accordingly. If no more steps are needed and you can return to the user, then respond with that. Otherwise, fill out the plan. Only add steps to the plan that still NEED to be done. Do not return previously done steps as part of the plan.
Answer in Korean."""
)
# Replanner 생성
replanner = replanner_prompt | ChatOpenAI(
model=MODEL_NAME, temperature=0
).with_structured_output(Act)
5. 노드 정의
그렇다면 LangGraph를 이용해 그래프를 만들기 위해 노드를 만드는 것이 필요하다.
from langchain_core.output_parsers import StrOutputParser
# 사용자 입력을 기반으로 계획을 생성하고 반환
def plan_step(state: PlanExecute):
plan = planner.invoke({"messages": [("user", state["input"])]})
# 생성된 계획의 단계 리스트 반환
return {"plan": plan.steps}
# 에이전트 실행기를 사용하여 주어진 작업을 수행하고 결과를 반환
def execute_step(state: PlanExecute):
plan = state["plan"]
# 계획을 문자열로 변환하여 각 단계에 번호를 매김
plan_str = "\n".join(f"{i+1}. {step}" for i, step in enumerate(plan))
task = plan[0]
# 현재 실행할 작업을 포맷팅하여 에이전트에 전달
task_formatted = f"""For the following plan:
{plan_str}\n\nYou are tasked with executing [step 1. {task}]."""
# 에이전트 실행기를 통해 작업 수행 및 결과 수신
agent_response = agent_executor.invoke({"messages": [("user", task_formatted)]})
# 이전 단계와 그 결과를 포함하는 딕셔너리 반환
return {
"past_steps": [(task, agent_response["messages"][-1].content)],
}
# 이전 단계의 결과를 바탕으로 계획을 업데이트하거나 최종 응답을 반환
def replan_step(state: PlanExecute):
output = replanner.invoke(state)
# 응답이 사용자에게 반환될 경우
if isinstance(output.action, Response):
return {"response": output.action.response}
# 추가 단계가 필요할 경우 계획의 단계 리스트 반환
else:
next_plan = output.action.steps
if len(next_plan) == 0:
return {"response": "No more steps needed."}
else:
return {"plan": next_plan}
# 에이전트의 실행 종료 여부를 결정하는 함수
def should_end(state: PlanExecute):
if "response" in state and state["response"]:
return "final_report"
else:
return "execute"
final_report_prompt = ChatPromptTemplate.from_template(
"""You are given the objective and the previously done steps. Your task is to generate a final report in markdown format.
Final report should be written in professional tone.
Your objective was this:
{input}
Your previously done steps(question and answer pairs):
{past_steps}
Generate a final report in markdown format. Write your response in Korean."""
)
final_report = (
final_report_prompt
| ChatOpenAI(model=MODEL_NAME, temperature=0)
| StrOutputParser()
)
def generate_final_report(state: PlanExecute):
past_steps = "\n\n".join(
[
f"Question: {past_step[0]}\n\nAnswer: {past_step[1]}\n\n####"
for past_step in state["past_steps"]
]
)
response = final_report.invoke({"input": state["input"], "past_steps": past_steps})
return {"response": response}
여기서 만들어지는 노드는 총 5개로 계획 노드, 실행 노드, 재계획 노드, 종료 시점 결정 노드, 최종 리포트 생성 노드이다.
개인적으로 주목하면 좋을 것 같다고 생각하는 노드는 종료 시점 결정 노드이다. 계획과 실행, 재계획을 반복하다보면 어디까지가 그 과정이 반복되야될지 명확하지 않아, 해당 과정을 계속 반복하게 될 수 도 있기 때문이다.
따라서 종료 시점을 어떻게든 결정할 수 도 있도록 LLM을 통해서든 rule base를 이용해서든 장치를 마련하는 것이 필요하다.
6. 그래프 생성 및 실행
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
# 그래프 생성
workflow = StateGraph(PlanExecute)
# 노드 정의
workflow.add_node("planner", plan_step)
workflow.add_node("execute", execute_step)
workflow.add_node("replan", replan_step)
workflow.add_node("final_report", generate_final_report)
# 엣지 정의
workflow.add_edge(START, "planner")
workflow.add_edge("planner", "execute")
workflow.add_edge("execute", "replan")
workflow.add_edge("final_report", END)
# 조건부 엣지: replan 후 종료 여부를 결정하는 함수 사용
workflow.add_conditional_edges(
"replan",
should_end,
{"execute": "execute", "final_report": "final_report"},
)
# 그래프 컴파일
app = workflow.compile(checkpointer=MemorySaver())
from langchain_teddynote.messages import invoke_graph, random_uuid
from langchain_core.runnables import RunnableConfig
config = RunnableConfig(recursion_limit=50, configurable={"thread_id": random_uuid()})
inputs = {
"input": "Modular RAG 가 기존의 Naive RAG 와 어떤 차이가 있는지와 production level 에서 사용하는 이점을 설명해줘"
}
invoke_graph(app, inputs, config)
노드를 연결하고, 실행하면 완료!
7. 함께 읽어보면 좋은 논문과 후기
해당 게시물을 읽고 습득하는 과정에서 다음 페이지와 논문들을 함께 읽으면 좋다.
https://github.com/yoheinakajima/babyagi
https://arxiv.org/abs/2305.04091
https://arxiv.org/abs/2210.03629
이 중 babyagi는 작년에 돌려본 경험이 있고, 테디 노트님도 해당 강의를 만드시면서 이 내용을 참고하셨다는 사실을 말씀하셨다. 개인적으로 해당 방법론의 가장 중요한 포인트는 출구 전략을 세우는 것이라고 생각한다.
계획을 세우고, 재계획을 수립하는 과정에서, 장기적인 목표를 보게되는데, babyagi를 돌려봤을 때의 경험을 다시 생각해보면 재계획을 반복하면 계속 세부적인 전략을 제안한다. 물론 세부적인 계획을 계속 만다는 것은 좋지만, 어디까지 세밀한 계획이 만들어져야하는지, 그 가이드라인이 명확하지 않으면, 다음 에이전트는 오랜 시간 토큰을 소비하면서 종료하지 않을 가능성이 높다.
따라서 시스템의 효율성을 위해 목적에 맞는 출구 전략을 꼭 마련해야 할 것 같다.
프로그래밍 기록 + 공부 기록