1. Introduction

본 내용에서는 앞서 다룬 Reflection Agent를 한 단계 발전시켜, LLM이 자체적으로 답변을 평가하고 개선하는 메타-프롬프트 워크플로우를 지닌 Reflexion Agent를 다룬다.

최근 LLM 기반 Agents은 단순히 입력에 반응하는 수준을 넘어서, 생성된 결과물을 스스로 점검하고 보강하여 성능과 일관성을 크게 향상시키고 있다. Reflection Agent가 “내가 만든 답변에 어떤 정보가 빠졌는가?”, “어디가 불필요하게 장황한가?”와 같이 엄정하게 비판하는 역할을 수행했다면, Reflexion Agent는 그 다음 단계로, 비판에서 도출된 개선점을 실제 답변에 반영하여 새로운 산출물을 생성한다.

논문 “Reflexion: Language Modeling via Verbal Reinforcement”에서는 전통적 강화학습 대신, 언어적 피드백을 통해 LLM이 반복적으로 학습·개선하도록 설계된 패러다임을 제안한다. 이 과정은 다음과 같이 크게 세 가지 구성 요소로 이루어진다.

• Actor: 질문에 대한 초기 답변 생성

• Reflection: 생성된 답변을 엄정히 비판하여 개선 여지를 도출

• Revision: 비판 결과를 바탕으로 답변을 보강·정제

2. Actor Agent

가. Actor Agent?

Actor Agent는 Reflexion Agent Architecture의 출발점이라 할 수 있다. 이 Component가 담당하는 첫 번째 미션은 사용자의 질문에 대해 전문 연구자 페르소나로서 250 단어 내외의 상세 답변을 작성하는 것이다. 이렇게 생성된 초기 결과는 이후의 평가와 개선 단계를 위한 기반이 된다.

하지만 단순한 답변 생성에 그치지 않는다. Actor Agent는 스스로 내 답변이 충분히 정확한가?, 어디가 더 보강되어야 할까? 라고 성찰하도록 설계되어 있다.

시스템 메시지에 "Reflect and critique your answer. Be severe to maximize improvement." 라는 지침을 포함하여 스스로 답변을 비판하고 개선 여지를 찾아내는 메타-사고 과정을 수행한다.

마지막으로, 답변의 빈틈을 채우기 위한 검색 쿼리까지 제안한다. “Recommend search queries to research information and improve your answer.”라는 단계가 그 역할로, 부족한 정보를 보충하기 위해 외부 지식 탐색 도구를 자동으로 연계할 수 있게 돕는다.

이로써 한 번의 질의가 “답변 → 자기 비판 → 추가 리서치”라는 유기적인 학습 루프를 돌며 점진적으로 완성도를 높이게 된다.

이처럼 Actor Agent는 질문에 대한 초기 대응뿐 아니라, 그 답변을 스스로 점검하고 보완 방향까지 제시함으로써 전체 Reflexion Agent Workflow의 질적 수준을 끌어올리는 핵심 모듈이다.

나. Code

1) Code for Chains

가) 환경 로드 및 import

# chains.py

import datetime
from dotenv import load_dotenv
load_dotenv()

from langchain_core.output_parsers.openai_tools import (
    JsonOutputToolsParser,
    PydanticToolsParser,
)
from langchain_core.messages import HumanMessage
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_openai import ChatOpenAI

from schemas import AnswerQuestion

나) LLM 및 parser 초기화

# chains.py

llm = ChatOpenAI(model="o4-mini")

# JSON 형태의 도구 호출 결과 처리
parser = JsonOutputToolsParser(return_id=True)

# Pydantic Schema 기반 유효성 검사를 수행
parser_pydantic = PydanticToolsParser(tools=[AnswerQuestion])

다) Actor Prompt Template 정의

# chains.py

actor_prompt_template = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            """You are expert researcher.
Current time: {time}

1. {first_instruction}
2. Reflect and critique your answer. Be severe to maximize improvement.
3. Recommend search queries to research information and improve your answer.""",
        ),
        MessagesPlaceholder(variable_name="messages"),
        ("system", "Answer the user's question above using the required format."),
    ]
).partial(
    time=lambda: datetime.datetime.now().isoformat(),
)

첫 번째 시스템 메시지에 세 가지 단계(초기 답변, 자기 비판, 추가 리서치 쿼리 추천)를 묶어 두고, {time} Placeholder에 현재 시각을 ISO 포맷으로 채워넣는다. 이는 현재 시점에 맞춘 최신 정보나 통계를 언급하여 시의성을 반영하기 위함이다.

라) First Responder Prompt 고정

# chains.py

first_responder_prompt_template = actor_prompt_template.partial(
    first_instruction="Provide a detailed ~250 word answer."
)

"약 250 단어 분량의 상세한 답변을 제공하라."라는 구체적인 지침을 넣어, 초기 답변 생성 역할을 완성한다.

마) Chain 구성

# chains.py

first_responder = first_responder_prompt_template | llm.bind_tools(
    tools=[AnswerQuestion], tool_choice="AnswerQuestion"
)

바) Script 실행 흐름

# chains.py

if __name__ == "__main__":
    human_message = HumanMessage(
        content="Write about AI-Powered SOC / autonomous soc  problem domain,"
        " list startups that do that and raised capital."
    )
    chain = (
        first_responder_prompt_template
        | llm.bind_tools(tools=[AnswerQuestion], tool_choice="AnswerQuestion")
        | parser_pydantic
    )

    res = chain.invoke(input={"messages": [human_message]})
    print(res)

HumanMessage로 사용자 질문을 래핑하여 messages 리스트에 담고, "Prompt → LLM + 도구 호출 → Pydantic Parsing" 순서로 Chain을 연결한다.

invoke을 호출하면 정해진 순서에 따라,

• 250 단어 답변 생성
• 답변에 대한 스스로의 반성 및 비판
• 추가 리서치용 검색 쿼리 추천
• 최종적으로 Pydantic Schema에 맞춘 구조화된 JSON 응답

위 모든 과정이 한 번에 실행되어 반환된다.

2) Code for Schema

가) Reflection 모델

# schema.py

from typing import List
from pydantic import BaseModel, Field


class Reflection(BaseModel):
    missing: str = Field(description="Critique of what is missing.")
    superfluous: str = Field(description="Critique of what is superfluous")

Reflection 모델은 초기 답변에 대한 reflection 정보를 구조화하기 위해 설계된 Pydantic 데이터 클래스이다. 핵심 필드는 두 가지이다.

• missing: str

  • 설명: 답변에서 빠진 부분을 기술하는 문자열

  • 용도: 사용자 질문에서 제대로 다루지 못했거나 더 보강해야 할 내용을 명확하게 지적

• superfluous: str

  • 설명: 답변 중 불필요하거나 과도한 부분을 기술하는 문자열

  • 용도: 핵심과 무관하게 삽입된 정보나 간결성을 해치는 여분의 내용을 짚어 준다.

위 두 속성을 통해, Actor Agent는 초기 답변을 빠진 점과 불필요한 점으로 나누어 비판하고, 이후 개선 작업에 구체적인 가이드를 제공할 수 있다.

나) AnswerQuestion 모델

# schemas.py

class AnswerQuestion(BaseModel):
    """Answer the question."""

    answer: str = Field(description="~250 word detailed answer to the question.")
    reflection: Reflection = Field(description="Your reflection on the initial answer.")
    search_queries: List[str] = Field(
        description="1-3 search queries for researching improvements to address the critique of your current answer."
    )

• answer:
  사용자 질문에 대한 약 250 단어 분량의 상세 답변을 담는 문자열 필드

• reflection:
  위에서 정의한 Reflection type으로, 답변에 대한 self-reflection(빠진 점, 불필요한 점)을 구조화하여 포함

• search_queries:
  답변을 보완하기 위해 실제로 수행할 검색 쿼리 목록(1~3 개)을 문자열 리스트 형태로 제안

3. Revisor Agent

가. Revisor Agent?

Revisor Agent는 “생성 → 반성 → 개선” 메타 워크플로우의 마지막 단계를 담당하는 모듈이다. Actor Agent가 만든 초기 답변과 Reflection 모델이 도출한 비판 내용을 입력받아, 빠진 정보를 보강하고 불필요한 부분을 제거하여 최종 답변의 완성도를 끌어올린다.

나. Code

1) 수정 지침 문자열 정의

# chains.py
...
from schemas import AnswerQuestion, ReviseAnswer
...
revise_instructions = """Revise your previous answer using the new information.
    - You should use the previous critique to add important information to your answer.
        - You MUST include numerical citations in your revised answer to ensure it can be verified.
        - Add a "References" section to the bottom of your answer (which does not count towards the word limit). In form of:
            - [1] https://example.com
            - [2] https://example.com
    - You should use the previous critique to remove superfluous information from your answer and make SURE it is not more than 250 words.
"""

위 문자열은 어떤 방식으로 답변을 수정할지에 대해 구체적인 단계별 지침을 담고 있다. 핵심 요구 사항은 다음과 같다.

• 기본 비판(missing)을 반영하여 빠진 정보를 보강

• 숫자 인용([1], [2], ...)을 반드시 삽입

• 답변 끝에 "References" 섹션 생성

• 불필요한 부분 제거

• 250 단어 이내 유지

2) Revisor Chain 정의

# chains.py

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

if __name__ == "__main__":
    human_message = HumanMessage(

• actor_prompt_template.partial(...)

  • 앞서 Actor Agent에서 정의한 actor_prompt_template의 {first_instruction} 자리에 revise_instructions을 덮어씌운다.

  • 이를 통해 Revision 전용 Prompt가 되는 새로운 Template을 얻는다.

• llm.bind_tools(...)

  • ReviseAnswer 스키마를 tool로 바인딩하여, LLM 호출 결과가 ReviseAnswer 형태의 구조화된 JSON으로 파싱되도록 한다.

3) ReviseAnswer 스키마 확장

# schemas.py

...

class ReviseAnswer(AnswerQuestion):
    """Revise your original answer to your question."""

    references: List[str] = Field(
        description="Citations motivating your updated answer."
    )

기존 AnswerQuestion 모델(answer, reflection, search_queries)을 상속받아, 추가로 references 필드를 선언해 “revision 결과에 포함된 인용 URL 목록”을 담도록 한다.

4. Tool Executor

가. 환경 로드

# tool_executor.py

from dotenv import load_dotenv
load_dotenv()

나. 검색 엔진 초기화

# tool_executor.py

from langchain_tavily import TavilySearch
tavily_tool = TavilySearch(max_results=5)

TavilySearch를 통해 외부 검색(예: 웹, 문서, 데이터베이스)을 실행할 수 있는 클라이언트를 생성한다. 여기서는 최대 5 개의 결과만 가져오도록 설정했다.

다. 검색 쿼리 실행 함수 정의

# tool_executor.py

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

search_queries 리스트(“키워드1”, “키워드2”…)를 받아 내부적으로 TavilySearch.batch() 메서드에 { "query": ... } 형태로 넘겨 일괄 검색을 수행한다. 이후 검색 결과(예: 문서 요약, 링크, 메타데이터 등)를 반환한다.

라. ToolNode 구성

# tool_executor.py

from langchain_core.tools import StructuredTool
from langgraph.prebuilt import ToolNode
from schemas import AnswerQuestion, ReviseAnswer

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

• StructuredTool.from_function(...)

  • run_queries 함수를 LangGraph가 이해할 수 있는 tool로 래핑

  • name: 어떤 schema(AnswerQuestion, ReviseAnswer)의 search_queries 필드를 처리할지 지정

• ToolNode

  • 여러 개의 StructuredTool을 모아서 하나의 노드(실행 유닛)로 묶어 준다.

  • 나중에 Message Graph 상에서, 이 노드를 만나면 해당 도구들을 자동으로 호출하여 검색 결과를 삽입하도록 연결할 수 있다.

마. 목적

이 코드 덕분에, Actor/Revisor 단계에서 “검색 쿼리” 형태로 생성된 문자열 목록을 실제 검색(스니펫·링크·요약 등)으로 자동 전환할 수 있다. LangGraph 워크플로우에서 “질문 → 답변 → 비판 → 검색 실행 → 리비전” 같은 흐름을 구현할 때, 이 execute_tools Node를 통해 외부 데이터 조회 단계가 매끄럽게 통합되는 역할을 한다.

5. Main

가. import

# main.py

from typing import List

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

from chains import revisor, first_responder
from tool_executor import execute_tools

나. 반복 횟수

# main.py

MAX_ITERATIONS = 2

"검색 → Revision" 사이클을 몇 번 반복할지 설정하는 상수이다.

다. Builder

# main.py

builder = MessageGraph()

builder.add_node("draft", first_responder)
builder.add_node("execute_tools", execute_tools)
builder.add_node("revise", revisor)

• draft: 질문을 받아 초기 답변을 생성하는 first_responder

• execute_tools: 생성된 search_queries를 실제로 실행해 주는 execute_tools

• revise: 검색 결과와 피드백을 반영해 답변을 최종 수정하는 revisor

# main.py

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

• 조건 없이 Edge 설정: draft → execute_tools → revise

라. revise 노드 이후에 호출될 조건 함수

# main.py

def event_loop(state: List[BaseMessage]) -> str:
    count_tool_visits = sum(isinstance(item, ToolMessage) for item in state)
    num_iterations = count_tool_visits
    if num_iterations > MAX_ITERATIONS:
        return END
    return "execute_tools"

• state(메시지 이력)에서 ToolMessage 인스턴스 수를 센다

• 이 횟수가 MAX_ITERATIONS(2)보다 크면 END를 반환하여 그래프를 종료

• 그렇지 않으면 execute_tools를 반환해 “다시 검색 실행”으로 돌아가도록 함

마. 그래프 완성 및 컴파일

# main.py

builder.add_conditional_edges("revise", event_loop)
builder.set_entry_point("draft")
graph = builder.compile()

• 조건부 분기: revise 단계가 끝날 때마다 event_loop 함수를 호출해, 검색 실행 횟수가 제한(MAX_ITERATIONS)을 넘으면 종료(END), 그렇지 않으면 다시 execute_tools로 돌아가도록 설정

• 진입점 지정: 그래프 실행은 draft 노드(초기 답변 생성)에서 시작

• 컴파일: compile() 호출로, 선언한 node·edge를 실제로 실행 가능한 graph 객체로 만든다

바. 그래프 구조 시각화, 워크플로우 실행 및 결과 확인

# main.py

print(graph.get_graph().draw_mermaid())


res = graph.invoke(
    "Write about AI-Powered SOC / autonomous soc  problem domain, list startups that do that and raised capital."
)
print(res[-1].tool_calls[0]["args"]["answer"])
print(res)

<참고 자료>
https://arxiv.org/pdf/2303.11366
https://blog.langchain.com/reflection-agents/
https://github.com/emarco177/langgraph-course/commit/e6dd3fcca03200934025b3ffd652bb9f02bde3c8
https://github.com/emarco177/langgraph-course/commit/8c0b27b05d1ae825c1554309c23663bc2937e22a
https://github.com/emarco177/langgraph-course/commit/a092b97f018a151889d4e287733ef93559d3ff70#diff-ffc5e73ff4ecf5dfa1f6cbb0fec369bb950938e5f4b57515335f0fd5e9ade9ea
https://github.com/emarco177/langgraph-course/commit/bb79ec6f76bfe2c43e6ed275230915765ef40f4f