[LangGraph] 요구사항 연속적으로 수집하기 (prompt generation)

사용자 요구 사항 기반으로 메타 프롬프트 생성하기

이번에는 프롬프트를 생성하도록 돕는 챗봇을 만들어보자. 챗봇은 먼저 사용자로부터 요구사항을 수집한 뒤, 이를 바탕으로 프롬프트를 생성하고 사용자 입력에 따라 이를 수정한다. 이 과정은 두 개의 별도 State로 나뉘며, LLM이 State 전환 시점을 결정한다.

핵심은 반복적으로 사용자에게 질문을 하여 사전에 정의한 요구사항을 모두 수집한다는 점!

 

이 개념은 사용자로부터 수집해야 하는 정보가 필요한 서비스의 경우로 확장 가능하다.

 

예를 들어, LLM 기반 검색 시스템을 구축할 때 필요한 요구사항 수집 등에서 유용하게 사용될 것 같다. 만약 다양한 회의록을 검색하는 챗봇을 구축한다고 가정해보자. 이때 정확한 회의록을 찾기 위해서는 회의가 진행된 연도, 부서, 주제 등이 필요한데, 사용자는 아마 이러한 정보를 정확하게 지칭하지 않을 확률이 높다. 따라서, 3가지 요구사항을 수집할 때까지 반복적으로 사용자에게 질문하며 요구사항을 도출하고 이를 기반으로 검색을 수행한다면, 검색 시스템의 성능을 보다 높일 수 있다.

 

사용자 요구 사항 기반 프롬프트 생성기

이번 글에서는 필요한 요구 사항을 반복적으로 수집한 뒤, 모든 요구 사항을 수집했다면 다음 task로 프롬프트를 생성하는 서비스를 만들어보자.

 

프롬프트를 생성할 때 필요한 요구 사항을 먼저 정의하자.

1. 프롬프트의 목적 (What the objective of the prompt is)
2. 프롬프트 템플릿에 전달되는 변수 (What variables will be passed into the prompt template)
3. output에서 포함되지 않아야 하는, 작업에 대한 모든 제약 조건 (Any constraints for what the output should NOT do)
4. 출력이 반드시 준수해야 하는 모든 요구 사항 (Any requirements that the output MUST adhere to)

앞으로 만들 서비스는 프롬프트를 생성하기 위해 위 4가지 요구 사항을 수집해야 한다고 가정한다.

정보 수집 부분

먼저 사용자 요구 사항을 수집할 그래프 부분을 정의해 보자. 이 부분은 특정 system 메세지로 LLM을 호출하는 것으로 작동한다. 그리고 만약 모든 요구 사항을 수집했을 때, 비로소 tool을 호출할 수 있다.

 

이 부분은 모두 프롬프트로 제어한다. 아래를 보면, LLM이 수집해야 하는 요구사항 4개와 함께 마지막에 모든 정보를 수집했을 때 관련된 tool을 호출하라고 명시해두었다.

 

또한, LLM에 tool을 바인딩하는데 이때 tool은 Pydantic 모델로 요구사항 4가지를 각 변수가 담도록 한다.

bind_tools() 메서드는 LLM에게 tool의 schema를 넘겨주는 것으로, python 함수, Pydantic model, TypedDict class, 또는 Langchain tool object가 가능하다.

 

그러면, 앞으로 구현할 information_gather 노드가 조금 명확해진다. 프롬프트에는 4가지 요구 사항을 마음대로 추측하지 말고, 사용자로부터 수집하라고 작성했으며 이와 함께 모든 정보가 수집되었을 때 비로소 tool을 호출하라고 했다. 즉, information_gather 노드는 State에서 대화 이력이 담긴 messages를 입력 받고, 만약 대화 이력에서 4가지 요구 사항을 식별할 수 없다면 식별해야 하는 요구 사항에 대한 질문을 할 것이고, 모든 요구 사항이 식별되었다면 tool을 호출한다. 이때 tool은 4개의 요구 사항이 담긴 pydantic 객체가 된다.

 

이후 상황도 예상해볼 수 있는데, tool_calls가 반환되면 해당 tool에 담긴 정보로 프롬프트를 생성할 것이다.

 

그럼 먼저 프롬프트 템플릿을 정의해보자.

from typing import List
from langchain_core.messages import SystemMessage
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field

# 템플릿 정의
template = """Your job is to get information from a user about what type of prompt template they want to create.

You should get the following information from them:

- What the objective of the prompt is
- What variables will be passed into the prompt template
- Any constraints for what the output should NOT do
- Any requirements that the output MUST adhere to

[IMPORTANT!]
- You must communicate in Korean. However, the prompt generation must be in English.
- If you are not able to discern this info, ask them to clarify! Do not attempt to wildly guess.
- After you are able to discern all the information, call the relevant tool.
"""

prompt = ChatPromptTemplate.from_messages(
    [("system", template), MessagesPlaceholder(variable_name="messages")]
)

 

위 템플릿을 받아, 요구한 4가지를 pydantic model로 정의하고 LLM에게 tool로 바인딩 후 정보를 수집하는 chain을 정의하자.

class PromptInstructions(BaseModel):
    """Instructions on how to prompt the LLM."""

    objective: str
    variables: List[str]
    constraints: List[str]
    requirements: List[str]


llm = ChatOpenAI(model="gpt-4o", temperature=0)
llm_with_tool = llm.bind_tools([PromptInstructions])

gather_chain = prompt | llm_with_tool

 

정의한 gather_chain으로 사용자로부터 프롬프트를 구성하기 위한 요구 사항을 수집하는 information_gather 노드를 아래와 같이 정의할 수 있다.

# information_gather 노드 정의
def information_gather(state):
    """사용자와 상호작용하며 필요한 요구 사항을 수집하는 노드"""
    messages = state["messages"]
    response = gather_chain.invoke({"messages": messages})
    return {"messages": [response]}

 

information_gather 노드가 잘 작동하는지 테스트해보자.

1. 단순히 프롬프트 생성을 요구하는 질문
   • 의도대로 4가지 정보를 요구하는 답변을 하고 있음
2. 4가지 요구 사항 중 제약 조건만을 포함한 질문
   • 4가지 요구 사항 중 나머지 3개에 대해 정보 요구
3. 4가지 요구 사항을 모두 포함한 질문
   • 4가지 요구 사항을 모두 알아냈기 때문에 tool_calls를 반환

## test
print(
    gather_chain.invoke(
        {"messages": [("user", "안녕하세요, 프롬프트 만들어주세요.")]}
    ).content
)
print(" --- ")
print(
    gather_chain.invoke(
        {
            "messages": [
                ("user", "안녕하세요, 프롬프트 만들어주세요. 제약 조건은 차별적인 발언을 하지 않는 것입니다.",)
            ]
        }
    ).content
)
print(" --- ")
print(
    gather_chain.invoke(
        {
            "messages": [
                (
                    "user",
                    "안녕하세요, 프롬프트 만들어주세요. 목적은 대화, 입력은 user_input, 제약이나 요구 사항은 없습니다.",
                )
            ]
        }
    )
)

안녕하세요! 프롬프트 템플릿을 만들기 위해 몇 가지 정보를 여쭤보겠습니다.

1. 프롬프트의 목표는 무엇인가요?
2. 프롬프트 템플릿에 어떤 변수가 포함될 예정인가요?
3. 출력 결과에서 피해야 할 제약 조건이 있나요?
4. 출력 결과가 반드시 따라야 할 요구 사항이 있나요?

이 정보를 알려주시면 프롬프트를 생성해드리겠습니다.
 --- 
안녕하세요! 프롬프트를 만들기 위해 몇 가지 정보를 더 필요로 합니다.

1. 프롬프트의 목표는 무엇인가요?
2. 프롬프트 템플릿에 어떤 변수가 포함될 예정인가요?
3. 출력 결과가 반드시 따라야 할 요구 사항이 있나요?

이 정보를 알려주시면 프롬프트를 생성하는 데 도움이 됩니다.
 --- 
content='' additional_kwargs={'tool_calls': [{'id': 'call_eYAZdUa1to1eYZjQWzGENGob', 'function': {'arguments': '{"objective":"Create a conversational prompt.","variables":["user_input"],"constraints":[],"requirements":[]}', 'name': 'PromptInstructions'}, 'type': 'function'}], 'refusal': None} response_metadata={'token_usage': {'completion_tokens': 31, 'prompt_tokens': 217, 'total_tokens': 248, 'completion_tokens_details': {'accepted_prediction_tokens': 0, 'audio_tokens': 0, 'reasoning_tokens': 0, 'rejected_prediction_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': 0, 'cached_tokens': 0}}, 'model_name': 'gpt-4o-2024-08-06', 'system_fingerprint': 'fp_4691090a87', 'finish_reason': 'tool_calls', 'logprobs': None} id='run-76b2cf12-c5a4-46ef-bc79-2d7553d20b0a-0' tool_calls=[{'name': 'PromptInstructions', 'args': {'objective': 'Create a conversational prompt.', 'variables': ['user_input'], 'constraints': [], 'requirements': []}, 'id': 'call_eYAZdUa1to1eYZjQWzGENGob', 'type': 'tool_call'}] usage_metadata={'input_tokens': 217, 'output_tokens': 31, 'total_tokens': 248, 'input_token_details': {'audio': 0, 'cache_read': 0}, 'output_token_details': {'audio': 0, 'reasoning': 0}}

 

이제 information_gather 노드는 그래프를 순회할 때, State에 담긴 대화 이력 messages를 받아 4가지 요구 사항을 모두 알아낼 때까지 질문을 던지게 된다. 그리고 4가지 요구 사항을 모두 획득했다면, tool_calls를 반환한다.

프롬프트 생성 노드

이번에는 프롬프트를 생성하는 generate_prompt 노드를 만들어보자.

 

generate_prompt 노드 직전에 information_gahter는 tool_calls를 반환했다. 따라서, generate_prompt 노드는 먼저 State의 마지막 AIMessage에 담긴 tool_calls를 식별하고, 해당 args를 인덱싱해서 프롬프트를 생성하는 데 필요한 4가지 정보들을 가져와야 한다.

 

이와 함께 프롬프트 생성을 위한 프롬프트를 생성하는 get_prompt_messages 함수를 정의하자.

• get_prompt_messages 함수
  • State의 messages에서 tool_calls가 포함된 AIMessage에서 tool_calls의 args를 가져옴
  • tool_calls 이후의 메세지를 가져와서 추가적인 대화 내용을 가져옴 (이는 프롬프트 생성 이후에 추가적으로 이루어진 대화 내역을 반영하기 위함)

from langchain_core.messages import AIMessage, HumanMessage, ToolMessage

generation_prompt = """Based on the following requirements, write a good prompt template:

{requirements}"""


def get_prompt_messages(messages: list):
    tool_call = None
    other_messages = []

    for msg in messages:
        if isinstance(msg, AIMessage) and msg.tool_calls:
            tool_call = msg.tool_calls[0]["args"]
        elif isinstance(msg, ToolMessage):
            continue
        # tool call 이후의 메세지 추가 (추가적인 대화 내용)
        elif tool_call is not None:
            other_messages.append(msg)
    return [
        SystemMessage(content=generation_prompt.format(requirements=tool_call))
    ] + other_messages

 

그리고 이 함수를 사용하여 LLM으로 프롬프트를 생성하는 prompt_generate 노드를 만들자.

llm = ChatOpenAI(model="gpt-4o", temperature=0)

def generate_prompt(state):
    messages = get_prompt_messages(state["messages"])
    response = llm.invoke(messages)
    return {"messages": [response]}

State 로직 정의

그래프를 구성하기 전, 마지막으로 State 로직을 정의하자. 즉, 사용자 질문의 information_gather 노드에 들어왔을 때, generate_prompt로 넘어가야할 지를 결정하는 조건부 함수를 구현한다.

• 마지막 메세지가 AIMessage고 tool_calls라면 종료하지 않고 tool_message를 추가하는 쪽으로
• 아니라면 종료

from langgraph.graph import END

def get_state(state):
    messages = state["messages"]
    if isinstance(messages[-1], AIMessage) and messages[-1].tool_calls:
        return "add_tool_message"
    return END

 

여기서 갑자기 add_tool_message 문자열을 반환하는 이유는, generate_prompt 노드로 가기 전 tool_calls가 실행되었다는 메세지를 추가해주기 위함이다. add_tool_message노드는 간단히 아래처럼 구현하자.

def add_tool_message(state):
    return {
        "messages": [
            ToolMessage(
                content="prompt 생성 완료!",
                tool_call_id=state["messages"][-1].tool_calls[0]["id"],
            )
        ]
    }

그래프 만들기

이제 그래프를 만들어보자.

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from typing import Annotated, TypedDict


class State(TypedDict):
    messages: Annotated[list, add_messages]


memory = MemorySaver()

flow = StateGraph(State)
flow.add_node("information_gather", information_gather)
flow.add_node("generate_prompt", generate_prompt)
flow.add_node("add_tool_message", add_tool_message)

flow.add_edge(START, "information_gather")
flow.add_conditional_edges(
    "information_gather",
    get_state,
    {
        "add_tool_message": "add_tool_message",
        END: END,
    },
)
flow.add_edge("add_tool_message", "generate_prompt")
flow.add_edge("generate_prompt", END)


graph = flow.compile(checkpointer=memory)

 

 

 

그래프 테스트

이제 그래프를 while 문을 돌면서 테스트해보자. 아래 코드로 config를 통해 State(특히 대화 내용)를 관리하면서 연속성 있게 대화를 이어나갈 수 있다.

 

결과를 보면, RAG 프롬프트를 요구하고 요구 사항이 만족되자 tool_calls를 반환해 프롬프트를 생성한다. 그리고 이후 프롬프트 수정사항을 이야기하면, 요구 사항을 반영해서 프롬프트를 수정해주는 걸 확인할 수 있다.

import uuid
import sys

thread_id = str(uuid.uuid4())
config = {"configurable": {"thread_id": thread_id}}

while True:
    user = input("User (q / Q to quit): ")

    print(f"User: {user}")

    if user in ["q", "Q"]:
        print("AI: 안녕히 계세요.")
        break

    output = None
    inputs = {"messages": [HumanMessage(content=user)]}
    for output in graph.stream(inputs, config, stream_mode="updates"):
        for k, v in output.items():
            if v is not None:
                print(
                    f"================================== Node: {k} =================================="
                )
                print(v["messages"][-1].pretty_print())
                sys.stdout.flush()
    if output and "generate_prompt" in output:
        print("Done!")

User: RAG 프롬프트
================================== Node: information_gather ==================================
================================== Ai Message ==================================

RAG 프롬프트 템플릿을 생성하기 위해 다음 정보를 제공해 주세요:

1. 프롬프트의 목표는 무엇인가요?
2. 프롬프트 템플릿에 어떤 변수가 전달되나요?
3. 출력에서 피해야 할 제약 조건이 있나요?
4. 출력이 반드시 따라야 할 요구 사항이 있나요?
None
User: 프롬프트의 목표는 블로그에 작성된 langgraph 관련 내용을 검색하고 설명해주는 것입니다. 변수로는 블로그 내용을 일부 가져온 context와 사용자 입력 user_input
================================== Node: information_gather ==================================
================================== Ai Message ==================================

출력에 대한 제약 조건이나 반드시 따라야 할 요구 사항이 있을까요? 예를 들어, 출력이 특정 형식을 따라야 한다거나, 특정 정보를 포함해서는 안 된다거나 하는 조건이 있을 수 있습니다.
None
User: 없어요.
================================== Node: information_gather ==================================
================================== Ai Message ==================================
Tool Calls:
  PromptInstructions (call_GkzDEgHCMY8gTq8eBTW3yIBS)
 Call ID: call_GkzDEgHCMY8gTq8eBTW3yIBS
  Args:
    objective: To search and explain content related to langgraph from a blog.
    variables: ['context', 'user_input']
    constraints: []
    requirements: []
None
================================== Node: add_tool_message ==================================
================================= Tool Message =================================

prompt 생성 완료!
None
================================== Node: generate_prompt ==================================
================================== Ai Message ==================================

**Prompt Template:**

"Based on the context provided, please search and explain content related to 'langgraph' from a blog. Use the following user input to guide your explanation:

Context: {context}

User Input: {user_input}

Ensure that your explanation is clear, concise, and informative, providing relevant insights or details about 'langgraph' as discussed in the blog."
None
Done!
User: 한국어로 대답하라는 문구를 추가하고 싶어요.
================================== Node: information_gather ==================================
================================== Ai Message ==================================
Tool Calls:
  PromptInstructions (call_jhSnv2d97jK5GPMo5KsdcAHw)
 Call ID: call_jhSnv2d97jK5GPMo5KsdcAHw
  Args:
    objective: To search and explain content related to langgraph from a blog, and respond in Korean.
    variables: ['context', 'user_input']
    constraints: []
    requirements: ['The response must be in Korean.']
None
================================== Node: add_tool_message ==================================
================================= Tool Message =================================

prompt 생성 완료!
None
================================== Node: generate_prompt ==================================
================================== Ai Message ==================================

**Prompt Template:**

"Based on the context provided, please search and explain content related to 'langgraph' from a blog. Use the following user input to guide your explanation:

Context: {context}

User Input: {user_input}

Ensure that your explanation is clear, concise, and informative, providing relevant insights or details about 'langgraph' as discussed in the blog. Please respond in Korean."
None
Done!
User: q
AI: 안녕히 계세요.

---

참고:
LangGraph. "Prompt Generation from User Requirements". https://langchain-ai.github.io/langgraph/tutorials/chatbots/information-gather-prompting/