1. Reflection이란?
Reflection은 에이전트가 자신의 결과물을 스스로 평가(자기 비판)하고,
필요하면 정해진 흐름으로 되돌아가 재실행하는 자기 개선 메커니즘입니다.
흐름은 크게 세 단계로 구성됩니다.
생성(초안 작성)
→ 반성(자기 비평 + 점수/비판 생성)
→ 수정(반성 내용을 반영해 재작성)이 순환이 반복되면서:
- 초안보다 정확하고 풍부한 결과물로 점점 개선되며
max_iters같은 반복 한도로 루프를 제어합니다.
2. 직접 구현해보는 Reflection – 가사 작성 예제
LangGraph 없이 프롬프트 체인만으로 Reflection 흐름을 직접 실습합니다.
목표: 이별 가사를 5단락으로 생성하고,
반성 피드백을 통해 더 나은 가사로 수정하는 흐름
2.1 생성 프롬프트 (generate)
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: 나중에 실제 대화 메시지들이 이 자리에 채워짐
MessagesPlaceholder(variable_name="messages"),
]
)
llm = ChatOpenAI(model="gpt-4o")
generate = prompt | llmChatPromptTemplate.from_messages: 메시지 목록으로 프롬프트 템플릿을 생성합니다.
MessagesPlaceholder: 동적으로 대화 메시지를 삽입할 자리를 예약합니다.
2.2 반성 프롬프트 (reflect)
reflection_prompt = ChatPromptTemplate.from_messages(
[
(
"system",
"당신은 노래 가사를 검토하고 개선 방향을 제안하는 작사 코치입니다. "
"사용자가 작성한 가사를 읽고 다음 형식으로 피드백을 작성하세요:\n"
"1. 잘한 점\n"
"2. 아쉬운 점\n"
"3. 구체적 개선 요청"
),
MessagesPlaceholder(variable_name="messages"),
]
)
reflect = reflection_prompt | llm2.3 Reflection 전체 흐름
request = HumanMessage(content='이별 노래 가사를 작성해줘')
# 1단계: 초안 생성
draft_lyrics = ""
for chunk in generate.stream({"messages": [request]}):
if chunk.content:
draft_lyrics += chunk.content
# 2단계: 피드백 생성
reflection_feedback = ""
for chunk in reflect.stream({
"messages": [request, HumanMessage(content=draft_lyrics)]
}):
if chunk.content:
reflection_feedback += chunk.content
# 3단계: 수정본 생성
revision_request = HumanMessage(
content=(
"아래 피드백을 반드시 반영하여 더 나은 가사를 작성해주세요.\n\n"
f"{reflection_feedback}"
)
)
for chunk in generate.stream({
"messages": [
request,
AIMessage(content=draft_lyrics), # 초안은 AI 메시지로
revision_request # 피드백 반영 요청
]
}):
if chunk.content:
print(chunk.content, end='')3. Graph로 Reflection 구현
위 흐름을 LangGraph 그래프로 자동화합니다.generate → reflect → generate → ... 루프가 메시지 수 기준으로 자동 종료됩니다.
3.1 State와 노드 정의
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import END, StateGraph, START
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"])]}reflection_node 에서는 메시지 타입을 변환해 줍니다.
def reflection_node(state: State) -> State:
cls_map = {"ai": AIMessage, "human": HumanMessage}
# reflection 프롬프트는 "사용자 요청 + AI 응답들" 형태를 기대하므로
# 기존 메시지의 타입을 반전시켜 전달
translated = [state["messages"][0]] + [
cls_map[msg.type](content=msg.content) for msg in state["messages"][1:]
]
response = reflect.invoke({"messages": translated})
# reflection 결과는 HumanMessage로 저장 (다음 generate 노드의 입력이 됨)
return {"messages": [HumanMessage(content=response.content)]}대화가 누적되면서 메시지 구조는 다음과 같이 쌓입니다.
[사용자 요청] → generation_node 입력
[사용자 요청] + [초안] → reflection_node 입력
[사용자 요청] + [초안] + [피드백] → generation_node 입력
[사용자 요청] + [초안] + [피드백] + [수정본] → reflection_node 입력
...3.2 조건부 종료 – should_continue
메시지가 6개를 넘으면 루프를 종료합니다.
from typing import Literal
def should_continue(state: State) -> Literal["reflect", END]:
if len(state["messages"]) > 6:
return END
return "reflect"3.3 그래프 조립 및 MemorySaver
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)MemorySaver: 그래프 실행 흐름을 메모리에 체크포인트로 저장합니다.
동일한 thread_id로 재실행하면 이전 상태를 이어서 사용할 수 있습니다.
3.4 실행
# 체크포인트를 사용하려면 config에 thread_id가 필요
config = {"configurable": {"thread_id": "1"}}
for event in graph.stream(
{"messages": [HumanMessage(content="이별 노래 가사를 작성해주세요.")]},
config,
):
print(event)
print("-" * 50)
# 최종 상태 확인
state = graph.get_state(config)
ChatPromptTemplate.from_messages(state.values["messages"]).pretty_print()4. Reflexion – 논문 기반 고도화 구현
Reflexion: Language Agents with Verbal Reinforcement Learning은
2023년 발표된 논문으로, 에이전트가 언어적 피드백(반성)을 장기 메모리에 저장하고
이를 다음 시도에 반영해 점진적으로 성능을 개선하는 프레임워크를 제안합니다.
논문의 핵심 구성 요소를 정리하면:
| 구성 요소 | 역할 |
|---|---|
| Actor (LM) | 실제 행동(답변 생성, 코드 작성 등)을 수행하는 언어 모델 |
| Evaluator (LM) | Actor의 결과를 평가해 점수/합/불합을 판단 |
| Self-Reflection (LM) | 평가 결과를 바탕으로 개선 방향을 담은 반성 텍스트를 생성 |
| Trajectory (단기 메모리) | 현재 시도의 행동/관찰 기록 |
| Experience (장기 메모리) | 이전 시도의 반성 내용을 누적해 다음 시도에 활용 |
| Environment | 코드 실행, 외부 검색 등 실제 피드백을 제공하는 환경 |
4.1 필요한 데이터 스키마 정의
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="답변에 대한 자기 반성")Responder 클래스는 체인을 감싸서 상태에서 입력을 받고 결과를 반환합니다.
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.2 초기 응답 체인 (Initial Responder)
import datetime
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)tool_choice="any": LLM이 반드시 도구를 호출하도록 강제합니다.
이를 통해 응답이 항상 AnswerQuestion 구조화 형식으로 반환됩니다.
4.3 수정 체인 (Revisor)
ReviseAnswer는 AnswerQuestion을 상속해 references 필드를 추가합니다.
class ReviseAnswer(AnswerQuestion):
"""Revise your original answer. Provide an answer, reflection,
cite your reflection with references, and add search queries."""
references: list[str] = Field(description="업데이트된 답변에 사용된 출처 목록")revise_instructions = """이전 답변을 새로운 정보로 수정하세요.
- 이전 비평에서 제기된 중요한 정보를 추가해야 합니다.
- 수정된 답변은 반드시 각주로 인용 표시를 포함해야 합니다.
- 답변 하단에 '참고문헌' 항목을 추가하세요.
- [1] https://example.com
- [2] https://example.com
- 불필요한 정보는 제거하고, 최종 답변은 반드시 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.4 검색 실행 도구 (ToolNode)
run_queries는 AnswerQuestion이 제안한 검색 쿼리를 Tavily로 일괄 실행합니다.
from langchain_core.tools import StructuredTool
from langgraph.prebuilt import ToolNode
def run_queries(search_queries: list[str], **kwargs):
"""Run the generated queries."""
# batch: 여러 개의 쿼리를 한 번에 실행하는 메서드
return tavily_tool.batch([{"query": query} for query in search_queries])
# AnswerQuestion, ReviseAnswer 두 도구 모두 동일한 run_queries 함수로 처리
tool_node = ToolNode(
[
StructuredTool.from_function(run_queries, name=AnswerQuestion.__name__),
StructuredTool.from_function(run_queries, name=ReviseAnswer.__name__),
]
)StructuredTool.from_function: 일반 Python 함수를 LangChain Tool 형식으로 변환합니다.
name을 Pydantic 모델 이름과 일치시켜서, LLM이 해당 도구를 호출할 때
run_queries가 실행되도록 연결합니다.
4.5 그래프 조립
from langgraph.graph import END, StateGraph, START
from langgraph.graph.message import add_messages
from typing import Annotated
from typing_extensions import TypedDict
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(state: list):
i = 0
# 메시지를 뒤에서부터 순회
# tool 또는 ai 메시지가 연속되는 구간이 한 번의 반복 사이클
for m in state[::-1]:
if m.type not in {"tool", "ai"}:
break
i += 1
return i
def event_loop(state: list):
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()전체 흐름을 정리하면:
START → draft (초안 + 검색 쿼리 생성)
→ execute_tools (검색 실행)
→ revise (검색 결과 반영해 수정)
→ (반복 횟수 초과?) → END
→ execute_tools (반복)4.6 실행
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 & Reflexion의 포인트
- Reflection은 "생성 → 자기 비평 → 수정"을 반복하며 결과물을 점진적으로 개선하는 패턴입니다.
- LangGraph에서는 generate → reflect → generate 루프를 그래프로 구성하고, 메시지 수 기준으로 종료 조건을 설정합니다.
- MemorySaver 를 checkpointer로 등록하면 thread_id를 통해 실행 상태를 저장·복원할 수 있습니다.
- Reflexion 논문은 Reflection을 고도화해 검색 도구와 결합하고, 답변에 출처와 반성을 구조화된 형식(AnswerQuestion, ReviseAnswer)으로 담아냅니다.
- StructuredTool.from_function 은 Python 함수를 LangChain Tool로 변환하며, name을 Pydantic 모델명과 일치시켜 LLM 도구 호출과 연결합니다.
- tavily_tool.batch 는 여러 검색 쿼리를 한 번에 실행하는 메서드로, 복수의 검색이 필요한 반성 루프에서 효율적으로 활용됩니다.