LangGraph Reflection - 자기평가로 답변을 개선하는 루프 설계

Reflection은 에이전트가 스스로 결과를 평가·비판한 뒤 피드백을 상태(state)에 기록하고, 필요하면 수정 루프로 되돌아가 답을 개선하는 설계 패턴이다. 보통 “작성 노드(답 생성) → 리플렉션 노드(자기평가) → 라우팅(조건부 엣지)” 구조로 만들고, 반복 횟수 제한(max_iters)을 둬 무한 루프를 막는다.

이번 글에서는

1) Reflection 개념
2) 랭체인 + OpenAI로 “가사 생성/평가/수정”
3) LangGraph로 Reflection 루프 구현
4) Reflexion(논문 아이디어) + 웹검색 툴 결합

까지 한 번에 정리한다.
원문 흐름은 랭그래프 Reflection 기반으로 재구성했다.

---

1. Reflection

Reflection의 핵심은 간단하다.

• 초안 생성
• 자기 평가(잘된 점/아쉬운 점/수정 지시)
• 피드백 반영해서 수정
• 기준 미달이면 다시 반복

즉, “한 번 생성하고 끝”이 아니라 “품질을 스스로 끌어올리는 루프”다.

import getpass
import os


def _set_env(var: str):
    if not os.environ.get(var):
        os.environ[var] = getpass.getpass(f"{var}: ")


_set_env("OPENAI_API_KEY")

!pip install langchain_openai

---

2. 랭체인 + OpenAI 모델을 이용한 가사 쓰기

여기서는 Reflection을 가장 직관적으로 확인하기 위해 “작사 도우미” 예제로 진행한다.

• 가사 생성기: 사용자 요청을 받아 5단락 가사 생성
• 가사 평가기: 잘된 점/아쉬운 점/수정 지시사항 생성
• 가사 수정기: 피드백을 반영해 개선본 생성

2-1. 생성 프롬프트

from langchain_core.messages import AIMessage, HumanMessage
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_openai import ChatOpenAI

prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "당신은 5단락 노래 가사를 작성하는 작사 도우미입니다. "
            "사용자의 요청에 맞게 주제, 분위기, 감정을 반영한 가사를 작성하세요. "
            "좋은 가사는 감정이 자연스럽게 이어지고, 표현이 구체적이며, 전체 흐름이 일관되어야 합니다. "
            "사용자가 피드백을 제공하면, 이전 결과를 그대로 반복하지 말고 "
            "지적된 부분이 눈에 띄게 개선된 수정본을 작성하세요. "
            "이때 잘된 표현과 핵심 주제는 유지하고, 감정 표현, 문장 자연스러움, 이미지 표현을 더 좋게 다듬으세요. "
            "반드시 5단락으로 작성하고, 각 단락은 명확히 구분하세요."
        ),
        MessagesPlaceholder(variable_name="messages"),
    ]
)

llm = ChatOpenAI(model="gpt-5.4-2026-03-05")
generate = prompt | llm

2-2. 평가 프롬프트

reflection_prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "당신은 노래 가사를 평가하고 개선 방향을 제안하는 작사 코치입니다. "
            "사용자가 제출한 가사를 읽고 주제의 일관성, 감정의 깊이, 표현의 구체성, "
            "문체의 자연스러움, 단락 간 흐름을 기준으로 평가하세요. "
            "응답은 반드시 다음 세 부분으로 작성하세요. "
            "1. 잘된 점 "
            "2. 아쉬운 점 "
            "3. 수정 지시사항 "
            "수정 지시사항은 다음 작사 단계에서 바로 반영할 수 있을 만큼 구체적으로 작성하세요."
        ),
        MessagesPlaceholder(variable_name="messages"),
    ]
)

reflect = reflection_prompt | llm

2-3. 초안 → 피드백 → 수정본

request = HumanMessage(content="이별에 대한 가사를 작성해줘.")

draft_lyrics = ""
for chunk in generate.stream({"messages": [request]}):
    if chunk.content:
        print(chunk.content, end="")
        draft_lyrics += chunk.content

reflection_feedback = ""
for chunk in reflect.stream({"messages": [request, AIMessage(content=draft_lyrics)]}):
    if chunk.content:
        print(chunk.content, end="")
        reflection_feedback += chunk.content

revised_lyrics = ""
revision_request = HumanMessage(
    content=(
        "아래 피드백을 반드시 반영하여 이전 가사를 수정해주세요. "
        "원문을 그대로 반복하지 말고, 표현과 감정을 더 풍부하게 개선해주세요.\n\n"
        f"{reflection_feedback}"
    )
)

for chunk in generate.stream(
    {"messages": [request, AIMessage(content=draft_lyrics), revision_request]}
):
    if chunk.content:
        print(chunk.content, end="")
        revised_lyrics += chunk.content

---

3. Graph로 Reflection 구현

위 과정을 그래프로 옮기면, 루프와 종료 조건을 더 깔끔하게 제어할 수 있다.

!pip install langgraph

from typing import Annotated, Literal
from typing_extensions import TypedDict

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

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


def generation_node(state: State) -> State:
    return {"messages": [generate.invoke(state["messages"])]}

def reflection_node(state: State) -> State:
    cls_map = {"ai": AIMessage, "human": HumanMessage}

    translated = [state["messages"][0]] + [
        cls_map[msg.type](content=msg.content) for msg in state["messages"][1:]
    ]
    res = reflect.invoke(translated)
    return {"messages": [HumanMessage(content=res.content)]}

def should_continue(state: State) -> Literal["reflect", END]:
    if len(state["messages"]) > 6:
        return END
    return "reflect"

graph_builder = StateGraph(State)
graph_builder.add_node("generate", generation_node)
graph_builder.add_node("reflect", reflection_node)
graph_builder.add_edge(START, "generate")
graph_builder.add_conditional_edges("generate", should_continue)
graph_builder.add_edge("reflect", "generate")

memory = MemorySaver()
graph = graph_builder.compile(checkpointer=memory)

config = {"configurable": {"thread_id": "1"}}

for event in graph.stream(
    {"messages": [HumanMessage(content="이별에 대한 가사를 작성해주세요.")]},
    config,
):
    print(event)
    print("-" * 50)

---

4. Reflexion 구현

Reflexion은 Reflection 아이디어를 더 체계화한 프레임워크다.
핵심은 “반성문(언어 피드백)을 메모리에 저장하고 다음 시도에 반영”한다는 점이다.
논문: Reflexion: Language Agents with Verbal Reinforcement Learning

4-1. 구성 요소

• Actor (LM): 실제 행동(답안 작성, 코드 생성 등) 수행
• Evaluator (LM): 결과 품질 평가
• Self-reflection (LM): 개선 피드백 생성
• Trajectory (short-term memory): 현재 시도의 행동/관찰 기록
• Experience (long-term memory): 누적된 반성문 저장
• Environment: 외부 보상/피드백(테스트, 툴 응답 등)

---

4-2. Tavily 검색 도구 준비

def _set_env(var: str):
    if not os.environ.get(var):
        os.environ[var] = getpass.getpass(f"{var}: ")


_set_env("TAVILY_API_KEY")

!pip install langchain_community langchain-tavily

from langchain_tavily import TavilySearch

tavily_tool = TavilySearch(max_results=5)

---

4-3. 데이터 클래스 정의

from pydantic import BaseModel, Field


class Reflection(BaseModel):
    missing: str = Field(description="누락되거나 부족한 부분에 대한 비평")
    superfluous: str = Field(description="불필요한 부분에 대한 비평")


class AnswerQuestion(BaseModel):
    answer: str = Field(description="질문에 대한 10문장 이내의 자세한 답변")
    search_queries: list[str] = Field(
        description="현재 답변 비평을 해결하기 위한 1~3개의 검색 쿼리"
    )
    reflection: Reflection = Field(description="답변에 대한 자기반성 내용")


class Responder:
    def __init__(self, runnable):
        self.runnable = runnable

    def respond(self, state: dict):
        response = self.runnable.invoke({"messages": state["messages"]})
        return {"messages": response}

---

4-4. 초기 답변기 / 수정 답변기

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

actor_prompt_template = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            """당신은 전문 연구자입니다.
1. {first_instruction}
2. <Reflect> 생성한 답변을 다시 되돌아보고 개선할 수 있도록 비판하세요.
3. <Recommend search queries> 답변 질을 높이기 위해 추가 조사할 웹 검색 쿼리를 추천하세요.
""",
        ),
        MessagesPlaceholder(variable_name="messages"),
        ("user", "\n\n<Reflect> 사용자 원래 질문과 지금까지의 행동을 되돌아보세요."),
    ]
)

initial_answer_chain = actor_prompt_template.partial(
    first_instruction="질문에 대한 10문장 이내의 자세한 답변을 제공해주세요."
) | llm.bind_tools(tools=[AnswerQuestion], tool_choice="any")

first_responder = Responder(runnable=initial_answer_chain)

class ReviseAnswer(AnswerQuestion):
    references: list[str] = Field(description="업데이트된 답변에 사용된 인용 출처")

revise_instructions = """이전 답변을 새로운 정보를 바탕으로 수정하세요.
- 이전 비평 내용을 활용해 중요한 정보를 추가하세요.
- 수정 답변에는 숫자 인용 표기를 포함하세요.
- 답변 하단에 참고문헌 섹션을 추가하세요.
- 불필요한 내용은 제거하고 최종 답변은 200자를 넘지 않게 작성하세요.
"""

revision_chain = actor_prompt_template.partial(
    first_instruction=revise_instructions
) | llm.bind_tools(tools=[ReviseAnswer], tool_choice="any")

revisor = Responder(runnable=revision_chain)

---

4-5. 웹 검색 ToolNode 생성

from langchain_core.tools import StructuredTool
from langgraph.prebuilt import ToolNode

def run_queries(search_queries: list[str], **kwargs):
    """Run the generated queries."""
    return tavily_tool.batch([{"query": query} for query in search_queries])

tool_node = ToolNode(
    [
        StructuredTool.from_function(run_queries, name=AnswerQuestion.__name__),
        StructuredTool.from_function(run_queries, name=ReviseAnswer.__name__),
    ]
)

---

4-6. 그래프 구성 + 반복 제어

from typing_extensions import TypedDict
from typing import Annotated
from langgraph.graph import END, START, StateGraph
from langgraph.graph.message import add_messages


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

MAX_ITERATIONS = 5

graph_builder = StateGraph(State)
graph_builder.add_node("draft", first_responder.respond)
graph_builder.add_node("execute_tools", tool_node)
graph_builder.add_node("revise", revisor.respond)

graph_builder.add_edge("draft", "execute_tools")
graph_builder.add_edge("execute_tools", "revise")

def _get_num_iterations(messages: list):
    i = 0
    for m in messages[::-1]:
        if m.type not in {"tool", "ai"}:
            break
        i += 1
    return i


def event_loop(state: State):
    num_iterations = _get_num_iterations(state["messages"])
    if num_iterations > MAX_ITERATIONS:
        return END
    return "execute_tools"

graph_builder.add_conditional_edges("revise", event_loop, ["execute_tools", END])
graph_builder.add_edge(START, "draft")
graph = graph_builder.compile()

events = graph.stream(
    {"messages": [HumanMessage(content="AI Agent가 무엇인가요?")]},
    stream_mode="values",
)

for i, step in enumerate(events):
    print(f"Step {i}")
    step["messages"][-1].pretty_print()

---

마치며

• Reflection은 에이전트 품질을 높이는 가장 실용적인 패턴 중 하나다.
• 초안 생성 → 평가 → 수정 루프만 잘 구성해도 결과 일관성이 크게 올라간다.
• LangGraph로 옮기면 반복/종료 조건, 체크포인트, 상태 추적이 명확해져 운영이 쉬워진다.
• Reflexion 방식까지 확장하면, 반성문과 검색 결과를 누적해 다음 시도를 더 똑똑하게 만들 수 있다.

---

참고

• 랭그래프 Reflection
• Reflexion 논문