[langgraph][Planning Agents] Plan and Execute

langgraph의 공식문서를 번역해 놓은 자료입니다. 필요한 경우 부연 설명을 추가하였습니다. 문제가 되면 삭제하겠습니다.

https://langchain-ai.github.io/langgraph/tutorials/plan-and-execute/plan-and-execute/

여기서는 "계획 및 실행" 스타일의 에이전트를 만드는 방법을 보여준다. 이는 Plan-and-Solve 논문과 Baby-AGI 프로젝트에서 크게 영감을 받았다.

핵심 아이디어는 먼저 여러 단계로 이루어진 계획을 세운 다음, 그 계획을 하나씩 실행해 나가는 것이다. 특정 작업을 완료한 후에는 계획을 다시 검토하고 필요한 경우 수정할 수 있다.

일반적인 계산 그래프는 다음과 같습니다.

이 방식은 한 번에 한 단계씩 생각하는 일반적인 ReAct 스타일 에이전트와 비교된다. "계획 및 실행" 스타일 에이전트의 장점은 다음과 같다.

명확한 장기 계획 수립 (심지어 매우 강력한 LLM도 장기 계획 수립에는 어려움을 겪을 수 있음)
실행 단계에서는 더 작고 약한 모델을 사용할 수 있고, 계획 단계에서는 더 크고 성능이 좋은 모델만 사용 가능

다음 예제는 LangGraph에서 이를 구현하는 방법을 설명한다. 에이전트 결과물은 다음 예시처럼 결과물이 생길것이다: (링크).

Setup

먼저, 필요한 패키지들을 설치하고 API 키를 설정해 보자.

$ pip install --quiet -U langgraph langchain-community langchain-openai tavily-python

다음으로 OpenAI API Key와 Tavily API를 만들자.

from dotenv import load_dotenv

load_dotenv()

Tools 정의

먼저 사용할 도구들을 정의하자. 이 간단한 예제에서는 Tavily를 통해 내장된 검색 도구를 사용할 것이다. 하지만, 직접 도구를 만드는 것도 매우 쉽다. 이를 만드는 방법에 대한 자세한 내용은 여기를 참고하라.

from langchain_community.tools.tavily_search import TavilySearchResults

tools = [TavilySearchResults(max_results=3)]

실행 에이전트 정의

이제 작업을 실행할 실행 에이전트를 생성하자. 이 예제에서는 각 작업에 동일한 실행 에이전트를 사용할 예정이지만, 반드시 이렇게 해야 하는 것은 아니다.

from langchain import hub
from langchain_openai import ChatOpenAI

from langgraph.prebuilt import create_react_agent

# 사용할 프롬프트를 가져온다. 수정해도 무관하다.
prompt = hub.pull("ih/ih-react-agent-executor")
prompt.pretty_print()

# Choose the LLM that will drive the agent
llm = ChatOpenAI(model="gpt-4o-mini")
agent_executor = create_react_agent(llm, tools, state_modifier=prompt)

================================ System Message ================================

You are a helpful assistant.

============================= Messages Placeholder =============================

{messages}

result = agent_executor.invoke({"messages": [("user", "한국시리즈 우승팀은 누구야?")]})
print(result)

{'messages': [HumanMessage(content='한국시리즈 우승팀은 누구야?', additional_kwargs={}, response_metadata={}, id='3e0f8cd7-f1b4-4ec8-9e74-256d5d3d2f17'), AIMessage(content='', additional_kwargs={'tool_calls': [{'id': 'call_k4K799WuK0J8hePjKFiNSXwo', 'function': {'arguments': '{"query":"한국시리즈 2023 우승팀"}', 'name': 'tavily_search_results_json'}, 'type': 'function'}], 'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 26, 'prompt_tokens': 95, 'total_tokens': 121, 'completion_tokens_details': {'audio_tokens': 0, 'reasoning_tokens': 0, 'accepted_prediction_tokens': 0, 'rejected_prediction_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': 0, 'cached_tokens': 0}}, 'model_name': 'gpt-4o-mini-2024-07-18', 'system_fingerprint': 'fp_0705bf87c0', 'finish_reason': 'tool_calls', 'logprobs': None}, id='run-102cc94b-6a86-4f54-8f3d-4d7fe98c2d43-0', tool_calls=[{'name': 'tavily_search_results_json', 'args': {'query': '한국시리즈 2023 우승팀'}, 'id': 'call_k4K799WuK0J8hePjKFiNSXwo', 'type': 'tool_call'}], usage_metadata={'input_tokens': 95, 'output_tokens': 26, 'total_tokens': 121, 'input_token_details': {'audio': 0, 'cache_read': 0}, 'output_token_details': {'audio': 0, 'reasoning': 0}}), ToolMessage(content='[{"url": "https://star8353.tistory.com/entry/프로야구-역대-우승팀-1982-2020", "content": "프로야구 한국시리즈 역대 우승팀 1982-2023 10팀이 약 7개월 동안 144경기를 펼치는 이유는 한국시리즈 우승을 위해서입니다. ... (생략)"}, {"url": "https://ccaboonda.tistory.com/442", "content": "[2023년 프로야구] 포스트시즌 : 한국시리즈 결과, 우승팀, MVP [2023년 프로야구] 포스트시즌 : 한국시리즈 결과, 우승팀, ... (생략)"}, {"url": "https://www.yna.co.kr/view/GYH20231113001500044", "content": "(서울=연합뉴스) 김영은 김민지 기자 = LG 트윈스가 13일 잠실구장에서 열린 ... (생략)"}]', name='tavily_search_results_json', id='4383c68b-458c-4941-af6f-106466f54445', tool_call_id='call_k4K799WuK0J8hePjKFiNSXwo', artifact={'query': '한국시리즈 2023 우승팀', 'follow_up_questions': None, 'answer': None, 'images': [], 'results': [{'title': '프로야구 한국시리즈 역대 우승팀 1982-2023 - 지식창고', 'url': 'https://star8353.tistory.com/entry/프로야구-역대-우승팀-1982-2020', 'content': '프로야구 한국시리즈 역대 우승팀 1982-2023 10팀이 약 7개월 동안 144경기를 펼치는 이유는 한국시리즈 우승을 위해서입니다. K... (생략)', 'score': 0.99978, 'raw_content': None}, {'title': '[2023년 프로야구] 포스트시즌 : 한국시리즈 결과, 우승팀, Mvp', 'url': 'https://ccaboonda.tistory.com/442', 'content': '[2023년 프로야구] 포스트시즌 : 한국시리즈 결과, 우승팀, MVP [2023년 프로야구] 포스트시즌 : 한국시리즈 결과, ... (생략)', 'score': 0.9992649, 'raw_content': None}, {'title': '[그래픽] 2023 프로야구 한국시리즈 결과 - 연합뉴스', 'url': 'https://www.yna.co.kr/view/GYH20231113001500044', 'content': '(서울=연합뉴스) 김영은 김민지 기자 = LG 트윈스가 13일 잠실구장에서 열린 2023 신한은행 ... (생략)', 'score': 0.9965301, 'raw_content': None}], 'response_time': 2.83}), AIMessage(content='2023년 한국시리즈 우승팀은 LG 트윈스입니다. LG 트윈스는 2023년 11월 13일에 열린 5차전에서 kt wiz를 6-2로 이기며 29년 만에 한국시리즈 정상에 올랐습니다.', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 65, 'prompt_tokens': 976, 'total_tokens': 1041, 'completion_tokens_details': {'audio_tokens': 0, 'reasoning_tokens': 0, 'accepted_prediction_tokens': 0, 'rejected_prediction_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': 0, 'cached_tokens': 0}}, 'model_name': 'gpt-4o-mini-2024-07-18', 'system_fingerprint': 'fp_0705bf87c0', 'finish_reason': 'stop', 'logprobs': None}, id='run-7e393aaf-6e9f-41fd-8c0d-39c2469a5fec-0', usage_metadata={'input_tokens': 976, 'output_tokens': 65, 'total_tokens': 1041, 'input_token_details': {'audio': 0, 'cache_read': 0}, 'output_token_details': {'audio': 0, 'reasoning': 0}})]}

상태 정의

이제 이 에이전트가 추적할 상태를 정의해 보자.

먼저, 현재 계획을 추적해야 한다. 이를 문자열 목록으로 표현한다.

다음으로, 이전에 실행된 단계를 추적해야 한다. 이를 (단계, 결과)로 구성된 튜플 목록으로 표현한다.

마지막으로, 최종 응답과 원래 입력을 나타낼 상태가 필요하다.

import operator
from typing import Annotated, List, Tuple
from typing_extensions import TypedDict


class PlanExecute(TypedDict):
    input: str
    plan: List[str]
    past_steps: Annotated[List[Tuple], operator.add]
    response: str

계획 단계

계획단계를 생각해보자. 계획을 생성하는데 함수 호출을 사용할 것이다.

from pydantic import BaseModel, Field


class Plan(BaseModel):
    """Plan to follow in future"""

    steps: List[str] = Field(
        description="different steps to follow, should be in sorted order"
    )

from langchain_core.prompts import ChatPromptTemplate

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.""",
        ),
        ("placeholder", "{messages}"),
    ]
)
planner = planner_prompt | ChatOpenAI(
    model="gpt-4o-mini", temperature=0
).with_structured_output(Plan)

planner.invoke(
    {
        "messages": [
            ("user", "현재 프로야구 우승팀 감독의 고향은 어디야?")
        ]
    }
)

steps=['2023년 프로야구 우승팀을 확인한다.', '우승팀의 감독 이름을 찾는다.', '감독의 고향 정보를 검색한다.', '감독의 고향을 확인하여 최종 답변을 작성한다.']

Re-Plan 단계

이전 단계의 결과를 기반으로 재계획을 하는 단계를 만들자.

from typing import Union


class Response(BaseModel):
    """Response to user."""

    response: str


class Act(BaseModel):
    """Action to perform."""

    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."""
)


replanner = replanner_prompt | ChatOpenAI(
    model="gpt-4o-mini", temperature=0
).with_structured_output(Act)

그래프 생성

from typing import Literal
from langgraph.graph import END


async 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 = await agent_executor.ainvoke(
        {"messages": [("user", task_formatted)]}
    )
    return {
        "past_steps": [(task, agent_response["messages"][-1].content)],
    }


async def plan_step(state: PlanExecute):
    plan = await planner.ainvoke({"messages": [("user", state["input"])]})
    return {"plan": plan.steps}


async def replan_step(state: PlanExecute):
    output = await replanner.ainvoke(state)
    if isinstance(output.action, Response):
        return {"response": output.action.response}
    else:
        return {"plan": output.action.steps}


def should_end(state: PlanExecute):
    if "response" in state and state["response"]:
        return END
    else:
        return "agent"

from langgraph.graph import StateGraph, START

workflow = StateGraph(PlanExecute)

# plan node 추가
workflow.add_node("planner", plan_step)

# execution step 추가
workflow.add_node("agent", execute_step)

# replan node 추가
workflow.add_node("replan", replan_step)

workflow.add_edge(START, "planner")

# plan에서 agent로 연결
workflow.add_edge("planner", "agent")

# agent에서 replan으로 연결
workflow.add_edge("agent", "replan")

workflow.add_conditional_edges(
    "replan",
    # 다음으로 어떤 노드를 호출할지 결정하는 함수를 전달한다.
    should_end,
    ["agent", END],
)

# 마지막으로, 컴파일한다.
# 이것은 LangChain Runnable로 컴파일된다.
# 이것은 다른 runnable처럼 사용할 수 있다.
app = workflow.compile()

from IPython.display import Image, display

display(
    Image(
        app.get_graph(xray=True).draw_mermaid_png(
            output_file_path="./plan_and_execute.png"
        )
    )
)

config = {"recursion_limit": 50}
inputs = {"input": "2011년 한국시리즈 MVP의 그 해 연봉은 얼마야?"}

async def run():
    async for event in app.astream(inputs, config=config):
        for k, v in event.items():
            if k != "__end__":
                print(v)


asyncio.run(run())

{'plan': ['2011년 한국시리즈 MVP 선수를 확인한다.', '확인한 MVP 선수의 2011년 연봉 정보를 찾는다.', '찾은 연봉 정보를 정리하여 최종 답변을 작성한다.']}
{'past_steps': [('2011년 한국시리즈 MVP 선수를 확인한다.', '2011년 한국시리즈 MVP 선수는 오승환(삼성 라이온즈)입니다.')]}
{'plan': ['오승환 선수의 2011년 연봉 정보를 찾는다.', '찾은 연봉 정보를 정리하여 최종 답변을 작성한다.']}
{'past_steps': [('오승환 선수의 2011년 연봉 정보를 찾는다.', '오승환 선수의 2011년 연봉 정보는 다음과 같습니다:\n\n- 2011년 오승환 선수는 연봉 2억 4000만원에 계약하였습니다. \n- 이는 전년(2010년) 연봉 2억 6000만원에서 2000만원이 삭감된 금액입니다. \n\n이 정보는 오승환 선수가 팔꿈치 수술을 받은 후의 연봉 삭감과 관련이 있습니다.')]}
{'response': '오승환 선수의 2011년 연봉은 2억 4000만원입니다.'}