LangGraph - Reflection Agent

hyeony·2025년 7월 22일

LLM langgraph

LLM

목록 보기

1/4

1. Creating the Reflector Chain and the Tweet Revisor Chain

가. Reflector Chain

1) Why Reflector Chain?

Reflector Chain은 단순히 트윗을 생성하는 것을 넘어, 모델 스스로 결과물을 평가하고 개선점을 찾아내는 메타-프롬프트 기법을 학습하기 위해 설계되었다.

여기서 Viral Twitter Influencer라는 구체적인 페르소나를 부여하여 트윗 길이 등 실전 지표에 맞춘 일관되고 깊이 있는 피드백을 확보할 수 있다.

이러한 평가 모듈을 첫 단계로 두면, 이후 Generator Chain과 결합하여 피드백 → 재생성 → 재평가의 반복적인 개선 워크플로우를 손쉽게 구현할 수 있다.

2) Prompt Template 정의

이 부분에서는 평가 기준이 담긴 텍스트를 시스템 메시지로 설정한다. 그 다음에 MessagesPlaceholder을 배치한다. 이 인스턴스는 런타임에 대화 메시지를 주입할 위치를 지정하는 역할을 한다.

# chain.py

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_openai import ChatOpenAI

reflection_prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You are a viral twitter influencer grading a tweet. Generate critique and recommendations for the user's tweet."
            "Always provide detailed recommendations, including requests for length, virality, style, etc.",
        ),
        MessagesPlaceholder(variable_name="messages"),
    ]
)

나. Generator Chain

1) Why Generator Chain?

Reflector Chain이 피드백을 제공한다면, Generator Chain은 그 피드백을 반영하거나 사용자의 요청에 맞춰 최고의 트윗을 작성하는 데 집중한다.

즉, 사용자가 Reflector Chain에서 받은 comment을 함께 넘기면, 그 지적 사항을 반영하여 트윗을 개선한다.

2) Prompt Template 정의

# chain.py

generation_prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You are a twitter techie influencer assistant tasked with writing excellent twitter posts."
            " Generate the best twitter post possible for the user's request."
            " If the user provides critique, respond with a revised version of your previous attempts.",
        ),
        MessagesPlaceholder(variable_name="messages"),
    ]
)

다. Chain Composition & LLM Connection

# chain.py

llm = ChatOpenAI(model="o4-mini")
generate_chain = generation_prompt | llm
reflect_chain = reflection_prompt | llm

2. Defining LangGraph Graph

가. 환경 설정 및 Import 등

# main.py

from typing import List, Sequence

from dotenv import load_dotenv
load_dotenv()

from langchain_core.messages import BaseMessage, HumanMessage
from langgraph.graph import END, MessageGraph

from chains import generate_chain, reflect_chain


REFLECT = "reflect"
GENERATE = "generate"

나. node 함수 정의

1) generation node

현재까지의 메시지 이력을 generate_chain으로 넘겨 새로운 트윗 생성 결과를 받는 래퍼함수(Wrapper Function)이다.

# main.py

def generation_node(state: Sequence[BaseMessage]):
    return generate_chain.invoke({"messages": state})

2) reflection node

메시지 이력을 reflect_chain에 넘겨 피드백 생성 결과(res.content)를 HumanMessage로 감싸 반환하는 함수이다.

# main.py

def reflection_node(messages: Sequence[BaseMessage]) -> List[BaseMessage]:
    res = reflect_chain.invoke({"messages": messages})
    return [HumanMessage(content=res.content)]

다. Graph Builder Setting

MessageGraph 인스턴스를 만들고, 두 개의 노드(generate, reflect)를 함수에 매핑한 뒤, entry point을 generate로 설정한다.

# main.py

builder = MessageGraph()
builder.add_node(GENERATE, generation_node)
builder.add_node(REFLECT, reflection_node)
builder.set_entry_point(GENERATE)

라. Branch Logic 정의

should_continue는 메시지 이력의 길이가 6 개를 넘으면 END(그래프 종료), 그렇지 않으면 reflect로 이동하는 Branch 함수이다.

# main.py

def should_continue(state: List[BaseMessage]):
    if len(state) > 6:
        return END
    return REFLECT

GENERATE Node에서 나가는 Edge을 should_continue 함수의 반환값에 따라 달리 연결한다. 즉, END이면 그래프를 종료시키고, REFLECT이면 generate → reflect로 이동한다.

# main.py

builder.add_conditional_edges(GENERATE, should_continue)

일반적인 edge 연결은 다음과 같이 한다. reflect node은 무조건 다시 generate node로 이어지는데, 즉, 한 번 평가가 끝나면 새로운 트윗 생성 단계로 돌아가도록 고정한 흐름이다.

# main.py

builder.add_edge(REFLECT, GENERATE)

마. Graph 컴파일, 시각화 그리고 실제 호출

# main.py

graph = builder.compile()
print(graph.get_graph().draw_mermaid())
graph.get_graph().print_ascii()

if __name__ == "__main__":
    print("Hello LangGraph")
    inputs = HumanMessage(content="""Make this tweet better:"
                                    @LangChainAI
            — newly Tool Calling feature is seriously underrated.

            After a long wait, it's  here- making the implementation of agents across different models with function calling - super easy.

            Made a video covering their newest blog post

                                  """)
    response = graph.invoke(inputs)