[하루 1시간] LangGraph로 스스로 평가하는 프롬프트 강화기 만들기

개발에 앞서...

LangGraph ?

State 형태로 프로세스를 구성할 수 있는 LangGraph를 발견... 이걸로 뭘 할 수 있을까?
원래는 Tools를 나열하고 프롬프팅으로 서브그래프를 만들고 수평적으로 확장하는게 일반적인 사용법 같다.
하지만 연습용이기 때문에 간단하지만 쓸만한, 그리고 생각해본걸 구현하려 한다.

Prompt 강화기?

시스템 프롬프트는 생각보다 매우 강력한 역할을 한다. 유저가 악의적인 목적을 갖지 못하게 할 수도 있으며,
출력 포멧에 대한 제한을 둘 수도 있고, 말투나 정책, 역할 등을 강력하게 제한 할 수 있다.
하지만 LLM은 중간 과정이 블랙박스이기 때문에, prompt x input = output 일 때, output이 answer가 아닐 수 있다.
가령, 나는 JSON형식으로 출력해주길 바라는데, "네, 알겠습니다! '''JSON 어쩌고... '''" 라고 대답하는 경우가 있다.
이러면 시스템 프롬프트를 잘 쓰는 사람이라면, 최면을 강하게 걸어서 매우 높은 확률로 JSON만 뽑아낼 수 있지만, 익숙하지 않다면 좌절하기 십상이다.

그래서 프롬프트를 방정식처럼 X x input = answer 이라면, X 값을 찾아주는건 어떨까? 하는 발상에서 시작했다.

읽으면 좋은 StateMachine (위키)

목표

LangGraph 기반으로 Gemini를 활용하여 프롬프트를 자동으로 강화하는 시스템을 구현한다.

아이디어 & 솔루션

입력(input), 초기 프롬프트(prompt_0), 정답(answer)을 세트로 하여 다음과 같은 과정을 거친다:

입력과 프롬프트로 출력을 생성
input x prompt_n = output_n
출력을 평가하여 점수와 피드백을 산출
output_n x estimate_prompt = (score_n, feedback_n)
점수가 임계값을 못넘으면 피드백으로 새 프롬프트 생성
if score_n < THRESHOLD: prompt_n x feedback_n x enhance_prompt = prompt_n+1 else: END

위 과정을 반복하여 원하는 수준의 프롬프트가 나올때까지 정제한다.

구현

사실 큰 그림은 아래 4번부터 1번으로 올라오는게 이해하기 좋다.
개발/스크립트의 순서에 따라 정의 부터 기술한다.

1. 상태 정의

State를 지나면서 현재 상태는 확정되지만, 이전에 State의 Node에서 리턴해준 값만 기억을 할 수 있다.
전역변수로 두고 작업하긴 뭐하니, 상태에 대한 값을을 통째로 들고다니는 EnhancerState 클래스를 하나 만들어준다.

class EnhancerState(TypedDict):
    count: int  # 현재 시도 횟수
    MAX_COUNT: int  # 최대 시도 횟수 
    threshold: int  # 목표 점수
    focus: str  # 평가 기준
    input_text: str  # 입력 텍스트
    initial_system_prompt: str  # 초기 프롬프트
    answer: str  # 정답 텍스트
    system_prompt_history: List[str]  # 프롬프트 히스토리
    output_history: List[str]  # 출력 히스토리
    score_history: List[float]  # 점수 히스토리 
    feedback_history: List[List[str]]  # 피드백 히스토리

history 자체를 다시 한 Epoch로 class로 만들어서 history: List[Epoch] 처럼 만드는게 더 깔끔할 것 같다.

2. 프롬프트 템플릿 정의

프롬프팅을 하면서 체득한 말 잘듣게 하는 방법

기본적으로 시스템 프롬프트에게는 역할을 알려주는게 좋다.
LLM 모델마다 다르지만, 대체로 맨 앞과 맨 뒤의 지시사항을 주로 따르고, 중간은 좀 잊어버리는 경향이 강한편이다.
그래서 맨 위에 역할을 부여해서 수많은 자아중 한가지를 골라주고, 마지막에 "너 이거는 좀 지켜야돼?" 하고 확실히 해주는 것이 강력한 시스템 프롬프팅이라 생각한다.

1) 평가 프롬프트

입력된 프롬프트로 생성해낸 출력물이 내가 원하는 정답인지를 LLM에게 다시 물어봐서 점수를 받는다.
프롬프트 내에서 특정 값을 뽑아내고 싶다면, 태그 혹은 <<>> 같은 특수한 기호들을 사용해서 표시하라고 명령하는 것이 좋다.
- 꼭 응답 전체가 XML이나 JSON일 필요는 절대 없다. 나중에 기술하겠지만, Regex로 뽑으면 되기 때문에 이런 방식을 선호한다.
LLM이 항상 내가 시키는대로 대답하진 않지만, 성능좋은 모델일 수록 일관되고 포맷에 맞춰 잘 출력한다.
~~평가 프롬프팅의 경우, 성능이 좋은 모델을 골라야지 구린 모델을 사용한다면 채점도 못하는 멍청한 선생님을 붙여주는 꼴이다.~~

estimate_template = ChatPromptTemplate.from_messages([
    SystemMessagePromptTemplate.from_template("""당신은 프롬프트 엔지니어 전문가입니다.
제시된 시스템프롬프트를 사용한 사용자의 질문에 대한 응답이 정답으로 제시된 목적에 얼마나 가까운지 0~100 사이의 점수로 평가하시오.
평가의 기준은 '{focus}' 입니다.
또한 제시된 시스템프롬프트의 어떤이유로 인해 응답이 의도한응답과 괴리가 생기는지 개선점을 리스트업해서 피드백하시오.

출력은 XML 형태로 합니다.

점수는 <score>INTEGER</score> 형태로, 피드백은 각 항목별로 <feedback>STRING</feedback> 형태로 출력하시오.

출력예시:
시스템 프롬프트 평가 점수는 <score>75</score> 입니다.
'{focus}'에 중점을 둔 평가에 의해 다음 개선점이 필요합니다.
<feedback_list>
<feedback>구어체가 아닌 문어체를 사용하시오.</feedback>
<feedback>전문성 있는 리스트업 형태가 아닌 서술형으로 작성하십시오.</feedback>
<feedback>사용자의 요청에 누락된 부분이 있습니다.</feedback>
</feedback_list>
"""),
    HumanMessagePromptTemplate.from_template("""
입력된 시스템프롬프트:
{prompt}

사용자 입력:
{query}

출력된 응답:
{output}

의도한 응답:
{answer}""")
])

2) 강화 프롬프트

강화 프롬프트는 위 피드백을 그대로 붙여넣어줘서 이전 프롬프트를 다시 만들 뿐이다.

enhance_template = ChatPromptTemplate.from_messages([
    SystemMessagePromptTemplate.from_template("""당신은 프롬프트 엔지니어 개선 전문가입니다.
사용자가 제시한 개선사항에 맞게 기존의 프롬프트를 개선하시오.
새롭게 개선된 프롬프트는 <enhanced_prompt>STRING</enhanced_prompt> 태그를 사용하여 구분하시오.

반드시 <enhanced_prompt> 태그가 응답에 포함되어야 합니다.

"""),
    HumanMessagePromptTemplate.from_template("""다음 지시를 참고하여 아래 프롬프트를 개선하세요.:
개선된 프롬프트는 <enhanced_prompt>STRING</enhanced_prompt>와 같이 태그를 사용하여 표기합니다.

개선사항:
{feedback}

프롬프트:
{prompt}
""")
])

3. 노드 함수 구현

1) generate_output(): 프롬프트로 출력 생성

원래 그냥 사용하는 llm.invoke 해서 출력물을 뽑아낸다.

def generate_output(state: EnhancerState) -> EnhancerState:
    if state['count'] == 0:
        state['system_prompt_history'].append(state['initial_system_prompt'])

    sys_prompt = state['system_prompt_history'][-1]

    messages = [
        SystemMessage(content=sys_prompt),
        HumanMessage(content=state['input_text'])
    ]
    response = llm.invoke(messages)
    state['output_history'].append(response.content)

    return state

2) estimate_score(): 출력 평가하여 점수와 피드백 생성

이전 프롬프트의 출력물을 평가하는 프롬프트를 실행한다.
~~1번과 다른 것은 없다.~~
간혹 맘에 안드는 출력이 생기는 경우 파싱오류가 생기기 때문에 3번까지는 재시도 하게 해준다.
- Graceful 하게 오류를 langgraph에서 해결해주지 않을까 했지만, 노드 내에서 발생하는 오류는 알아서 처리해야 하는것 같다.
- 물론 오류가 나면 오류 Node를 하나 만들어서 이쁘게 보내줘도 되지만, 어차피 해당 노드로 올 것이기도 하고, 작은 프로그램이기 대문에 노드 내에서 해결하게 했다.

def estimate_score(state: EnhancerState) -> EnhancerState:
    output = state['output_history'][-1]
    messages = estimate_template.format_messages(
        focus=state['focus'],
        prompt=state['system_prompt_history'][-1],
        query=state['input_text'],
        output=output,
        answer=state['answer']
    )

    TRY_COUNT = 3
    count = 1
    while count < TRY_COUNT:
      count += 1
      try:
        response = llm.invoke(messages)
        content = response.content
        score = re.findall(r"<score>(\d+)</score>", content)[0]
        feedback_list = re.findall(r"<feedback>(.*?)</feedback>", content)

        state['score_history'].append(float(score))
        state['feedback_history'].append(feedback_list)
      except Exception as e:
        print("ERROR:", e)
        if count == TRY_COUNT:
          raise e

    return state

3) enhance_prompt(): 피드백으로 새 프롬프트 생성

전달받은 피드백 넣고 다시 생성...

    messages = enhance_template.format_messages(
        feedback="\n".join(state['feedback_history'][-1]),
        prompt=state['system_prompt_history'][-1]
    )
    TRY_COUNT = 3
    count = 1
    while count < TRY_COUNT:
      count += 1
      try:
        response = llm.invoke(messages)
        enhanced_prompt = re.findall(r'<enhanced_prompt>([\s\S]*?)</enhanced_prompt>', response.content)[0]
        state['system_prompt_history'].append(enhanced_prompt)
        state['count'] += 1
        break
      except Exception as e:
        print("ERROR:", e)
        if count == TRY_COUNT:
          raise e

    return state

4. LangChain LLM

무료로 API를 주는 gemini-1.5-flash 모델을 사용한다.
api_key는 colab에서 설정하거나, 하드코딩 해도 된다.

import os
from google.colab import userdata

api_key = userdata.get('GOOGLE_API_KEY')

from langchain_google_genai import ChatGoogleGenerativeAI
# 모델 로드
llm = ChatGoogleGenerativeAI(
    model="gemini-1.5-flash",
    temperature=0,
    google_api_key=api_key,
)

4. 그래프 구성

위에 작성한 노드 함수를 간단하게 노드와 엣지로 표현해주면 된다.
시도횟수와 평가점수는 분기되어야 하기 때문에 add_conditional_edges를 사용한다는 것 빼면, 구성은 매우 쉽다.

workflow = Graph()
workflow.add_node("GENERATE", generate_output)
workflow.add_node("ESTIMATE", estimate_score) 
workflow.add_node("ENHANCE", enhance_prompt)

workflow.add_edge("GENERATE", "ESTIMATE")

def condition(state: EnhancerState):
    return state["score_history"][-1] < state["threshold"] and state["count"] < state["MAX_COUNT"]

workflow.add_conditional_edges("ESTIMATE", condition, {
    True: "ENHANCE",
    False: END
})

workflow.add_edge("ENHANCE", "GENERATE")
workflow.set_entry_point("GENERATE")

app = workflow.compile()

마지막으로 app.get_graph().draw_mermaid_png()로 그려진 워크플로우를 이미지로 볼 수 있다.
ipynb라면 IPython을 이용하자.

from IPython.display import Image, display
display(Image(app.get_graph().draw_mermaid_png()))

실행

초기 셋팅

간단한 뉴스기사를 넣으면, 구어체+채팅체로 바꿔주길 원하는 것을 가정했다.

new_state = EnhancerState(
    count=0,
    MAX_COUNT=5,
    threshold=80,
    focus="말투",
    initial_system_prompt="""입력을 텍스트를 다음 지시사항에 맞게 말투를 변경하시오.
- 재미있게 말해야한다.
""",
    input_text="""우크라이나 전쟁 1,000일을 맞았지만, 전쟁은 더 악화 국면으로 접어드는 모습입니다. 우크라이나가 미국이 제공한 미사일로 러시아 본토 타격을 강행했고, 러시아는 핵 공격 대상에 우크라이나를 포함했습니다.""",
    answer="헐.. 우크라이나랑 벌써 전쟁 1000일이나 했는데 계속 싸우네... 미국에서 준 미사일로 러시아 본토 때리니까 러시아는 핵쏜다고 하네;;",
    system_prompt_history = [],
    output_history = [],
    score_history = [],
    feedback_history = []
)

실행

실행은 그냥 app.run(state) 해도 되지만, 추적을 해야하기 때문에 스트림을 사용한다.

print("\n* Initial State --------------------")
pprint.pp(new_state)

for output in app.stream(new_state):
  print("\n* Stream --------------------")
  for k, v in output.items():
    print(f"NODE[{k}]-- count={v['count']}")
    if k == "GENERATE":
      pprint.pprint({
          "output": v["output_history"][-1],
      })
    elif k == "ESTIMATE":
      pprint.pprint({
          "score": v["score_history"][-1] if len(v["score_history"]) else None,
          "feedback": v["feedback_history"][-1] if len(v["feedback_history"]) else None,
      })
    elif k == "ENHANCE":
      pprint.pprint({
          "new_prompt": v["system_prompt_history"][-1],
      })
    else:
      pprint.pprint(v)

결과

만족스럽다.

자기 혼자 꿍짝거리면서 열심히 스스로 피드백 하고 개선해나가는 모습이다.

마치며...

머신러닝 할 때도 마찬가지지만, learning_rate를 너무 크게주면 오버슈팅 or 발산 해버리는 현상이 있다.
피드백 한 프롬프트가 이전 점수보다 낮다면 다시 _롤백 하는 State_를 추가하면 더 좋을 것 같다.
생각보다 토큰을 많이 잡아 먹기도 한다. ~~(비싸다)~~
입력한 초기 시스템 프롬프트나 피드백을 고려해서 max_token의 개수를 줄이는 것도 방법일것같다.

저작자표시 (새창열림)

'Dev' 카테고리의 다른 글

Git 마지막 커밋 Author, Commiter 바꾸기 (1)	2024.07.23

개발에 앞서...

LangGraph ?

Prompt 강화기?

목표

아이디어 & 솔루션

구현

1. 상태 정의

2. 프롬프트 템플릿 정의

3. 노드 함수 구현

4. LangChain LLM

4. 그래프 구성

실행

초기 셋팅

실행

결과

마치며...

'Dev' 카테고리의 다른 글

티스토리툴바