에이전트 아키텍처의 성능을 높이는 두 가지 방법

• 성찰(reflection) : 인간 사고 양식 중 하나를 본떠, LLM 애플리케이션이 기존 출력 및 선택 과정을 분석하고 이전 반복 단계 성찰을 기억
• 멀티 에이전트(multi-agent) : 개인보다 팀이 더 큰 성과를 거둘 수 있듯, 어떤 문제는 여러 LLM 에이전트가 집단으로 협력하면 쉽게 해결할 수 있다.

1. 성찰(Reflection)

성찰(자아비판)은 여러 프롬프트 작성 기법 중 지금까지 다루지 않은 기법으로, 창작 프롬프트와 수정 프롬프트 사이의 반복적인 과정을 통해 형성된다. 이 과정은 인간이 창조하는 과정을 모방한 결과다.

인간의 사고방식에는 시스템1(반응적 또는 본능적)과 시스템 2(체계적이며 성찰적) 라는 두 양상이 존재

성찰을 올바르게 활용할 경우 LLM 애플리케이션 시스템 2의 행동을 닮아가는 방향으로 발전하는데 기여할 수 있다.

성찰은 generate 노드와 reflect 노드로 구성된 그래프를 구현

• generate노드 : 에세이 초안을 작성하거나 개정
• reflect노드 : 이후 수정에 반영할 비평을 작성

[성찰 아키텍처 코드]

from typing import Annotated, TypedDict

from langchain_core.messages import (
    AIMessage,
    BaseMessage,
    HumanMessage,
    SystemMessage,
)
from langchain_ollama import ChatOllama
from langchain_openai import ChatOpenAI
from langgraph.graph import END, START, StateGraph
from langgraph.graph.message import add_messages

from config.load_sys import llama_model

model = ChatOllama(model=llama_model, temperature=0.1)

# 상태 타입 정의
class State(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]

# 프롬프트 정의
generate_prompt = SystemMessage(
    '당신은 훌륭한 3단락 에세이를 작성하는 임무를 가진 에세이 어시스턴트입니다.'
    '사용자의 요청에 맞춰 최상의 에세이를 작성하세요.'
    '사용자가 비평을 제공하면, 이전 시도에 대한 수정 버전을 응답하세요.'
)

reflection_prompt = SystemMessage(
    '당신은 에세이 제출물을 평가하는 교사입니다. 사용자의 제출물에 대해 비평과 추천을 생성하세요.'
    '길이, 깊이, 스타일 등과 같은 구체적인 요구사항을 포함한 자세한 추천을 제공하세요.'
)

def generate(state: State) -> State:
    answer = model.invoke([generate_prompt] + state['messages'])
    return { 'messages': [answer] }

def reflect(state: State) -> State:
    cls_map = {AIMessage: HumanMessage, HumanMessage: AIMessage}
    # 메시지들을 반전시켜 LLM이 자신의 출력을 성찰하도록 한다.
    translated = [reflection_prompt, state['messages'][0]] + [
        cls_map[msg.__class__](content=msg.content) for msg in state['messages'][1:]
    ]
    answer = model.invoke(translated)
    ## 출력 결과를 생성기(generator)에 대한 사용자 피드백으로 취급
    return {'messages': [HumanMessage(content=answer.content)]}

def should_continue(state: State):
    if len(state['messages']) > 6:
        # 3회 반복 후, 각 반복마다 2개의 메시지가 쌓이면 종료합니다.
        return END
    else:
        return 'reflect'

## 그래프 구축
builder = StateGraph(State)
builder.add_node('generate', generate)
builder.add_node('reflect', reflect)
builder.add_edge(START, 'generate')
builder.add_conditional_edges('generate', should_continue)
builder.add_edge('reflect', 'generate')

graph = builder.compile()

graph.get_graph().draw_mermaid_png(output_file_path='graph_reflection.png')

## 에이전트 실행
input = {
    'messages': [
        HumanMessage(
              content='오늘날 \'어린 왕자\'가 왜 중요한지에 대해 에세이를 작성하세요.'

        )
    ]
}
for output in graph.stream(input):
    message_type = 'generate' if 'generate' in output else 'reflect'
    print('\nNew message:', output[message_type]
          ['messages'][-1].content[:100], '...')

reflect 노드는 LLM이 비평하는 에세이가 사람이 썻다고 생각하게 우회한다. 동시에 generate 노드는 비평을 사용자가 썻다고 생각하게 우회한다.

우회하는 이유

• LLM이 인간과 AI의 메시지 쌍을 중심으로 대화형 파인튜닝을 거쳤기 때문
• 채팅모델은 사용자가 연속으로 다수의 메시지를 보낼 경우 학습한 형식과 맞지 않아 성능이 떨어지 위험이 있으므로, 오고 가는 형태로 데이터를 정리

New message: "어린 왕자"는 1943년 프랑스 작가 앙투안 드 생텍쥐페리( Antoine de Saint-Exupéry)가 쓴 소설로, 어린이와 성인 모두에게 의미 있는 작품입니다.

첫째,  ...

New message: **비평**

*   에세이는 주제를 잘 정의하고 있지만, 구체적인 예시와 분석을 더 포함할 필요성이 있습니다.
*   각 장을 독립적으로 평가하는 것이 좋습니다. 첫 번째 장은  ...

New message: "어린 왕자"는 1943년 프랑스 작가 앙투안 드 생텍쥐페리( Antoine de Saint-Exupéry)가 쓴 소설로, 어린이와 성인 모두에게 의미 있는 작품입니다.

2. 서브그래프(subgraphs)

서브그래프는 다른 그래프의 일부로 들어가는 하위 그래프를 의미한다. 랭그래프를 활용해 다중 에이전트 아키텍처를 실현하는 핵심 기술 개념

부모 그래프에 서브 그래프를 추가하는 방법 두 가지

• 서브그래프를 직접호출하는 노드 추가
  • 부모 그래프와 서브 그래프가 상태 키를 공유할 경우 사용하기 좋다.
  • 상태를 입출력할 때 별도의 변환 과정을 거칠 필요 없이 그대로 활용 가능
• 서브 그래프를 호출하는 함수를 포함한 노드를 추가
  • 부모 그래프와 서브 그래프가 서로 다른 상태 키를 사용할 때 좋다.
  • 서브 그래프를 호출하기 전후에 상태를 반환해야 한다.

from typing import TypedDict
from langgraph.graph import START, StateGraph

# 부모 그래프와 서브그래프에서 사용할 상태
class State(TypedDict):
    foo: str  # 서브그래프와 이 키를 공유

class SubgraphState(TypedDict):
    foo: str  # 부모 그래프와 이 키를 공유
    bar: str

# 서브그래프 정의
def subgraph_node(state: SubgraphState):
    # 서브그래프 노드는 공유 키인 "foo"를 사용해 부모 그래프와 통신한다
    return {"foo": state["foo"] + "bar"}

subgraph_builder = StateGraph(SubgraphState)
subgraph_builder.add_node("subgraph_node", subgraph_node)
subgraph_builder.add_edge(START, "subgraph_node")
# 서브그래프에 필요한 추가 설정은 여기에 작성
subgraph = subgraph_builder.compile()

# 부모 그래프 정의
builder = StateGraph(State)
builder.add_node("subgraph", subgraph)
builder.add_edge(START, "subgraph")
# 부모 그래프에 필요한 추가 설정은 여기에 작성
graph = builder.compile()

# 예시
initial_state = {"foo": "hello"}
result = graph.invoke(initial_state)
print(f"Result: {result}")  # foo에 "bar"가 추가되어야 함

Result: {'foo': 'hellobar'}

[부모 그래프가 키를 공유하지 않을 때]

from typing import TypedDict
from langgraph.graph import START, StateGraph

class State(TypedDict):
    foo: str

class SubgraphState(TypedDict):
    # 부모 그래프와 키를 공유하지 않음
    bar: str
    baz: str

# 서브그래프 정의
def subgraph_node(state: SubgraphState):
    return {"bar": state["bar"] + "baz"}

subgraph_builder = StateGraph(SubgraphState)
subgraph_builder.add_node("subgraph_node", subgraph_node)
subgraph_builder.add_edge(START, "subgraph_node")
# 서브그래프에 필요한 추가 설정은 여기에 작성
subgraph = subgraph_builder.compile()

# 서브그래프를 호출하는 부모 그래프 정의
def node(state: State):
    # 부모 그래프의 상태를 서브그래프 상태로 변환
    response = subgraph.invoke({"bar": state["foo"]})
    # 응답을 다시 부모 그래프의 상태로 변환
    return {"foo": response["bar"]}

builder = StateGraph(State)
# 서브그래프 대신 `node`를 지정
builder.add_node("node", node)
builder.add_edge(START, "node")
# 부모 그래프에 필요한 추가 설정은 여기에 작성
graph = builder.compile()

# 예시
initial_state = {"foo": "hello"}
result = graph.invoke(initial_state)
print(
    f"Result: {result}"
)  # foo를 bar로 변환해 "baz"를 추가하고 다시 foo로 변환

멀티 에이전트 아키텍처

LLM 에이전트의 크기와 범위, 복잡성이 증가할수록 문제가 발생할 가능성이 높아지며, 성능에 영향을 미칠 우려가 있다.

• 에이전트에 선택할 툴을 지나치게 많이 제공하면 이후 호출할 툴을 판단하기 어렵다.
• 컨텍스트 프롬프트 길이가 단일 에이전트가 파악하기 어려울 정도로 복잡해져 사용하는 모델의 능력을 초과
• 사용자가 특정분야(기획, 연구, 수학 문제 풀이)에 특화된 하위 시스템의 사용을 희망

이러한 문제를 해결하기 위해 애플리케이션을 여러 개의 소규모 독립 에이전트로 분할 후, 이들을 멀티 에이전트 시스템을 구성하는 방한을 강구할 수 있다.

• 독립에이전트는 프롬프트와 LLM 호출만으로 간단하게 구성하고, ReACT 에이전트처럼 복잡하게 구현할 수 있다.

• 네트워크 아키텍처
  • 각 에이전트가 다른 에이전트와 상호 통신한다. 모든 에이전트가 다음에 실행될 에이전트를 결정
• 슈퍼바이저 아키텍처
  • 모든 에이전트가 슈퍼바이저(감독자)라 부르는 하나의 에이전트와 통신. 슈퍼바이저 에이전트는 이후 호출할 에이전트를 결정. 경우에 따라 슈퍼바이저 에이전트를 툴로 호출하는 방식으로 구현
• 계층 아키텍처
  • 여러 슈퍼바이저를 총괄하는 하나의 슈퍼바이저를 활용해 멀티 에이전트 시스템을 정의한다. 이 아키텍처는 슈퍼바이저 아키텍처를 일반화해 복잡한 제어 흐름을 보다 정교하게 관리하도록 설계
• 맞춤형 멀티 에이전트 워크플로
  • 각 에이전트가 허용된 일부 노드와만 통신한다. 순서가 정해진 단계와 선택 가능한 단계가 섞일 수 있으므로, 어느 상태를 누구에게 넘길지 명시한다.

1. 슈퍼바이저 아키텍처

슈퍼바이저 아키텍처는 각 에이전트를 그래프의 노드로 추가함과 동시에 호출 순서를 결정하는 슈퍼바이저 노드를 배치. 조건부 엣지를 활용해 슈퍼바이저의 판단에 따라 적절한 에이전트 노드로 유도.

flowchart TD
    U[사용자 요청] --> S[슈퍼바이저]
    S -->|researcher| R[자료 조사]
    S -->|coder| C[구현 초안]
    S -->|FINISH 또는 단계 상한| E[종료]
    R --> S
    C --> S

슈퍼바이저는 다음 노드를 허용된 이름 중 하나로 선택한다. 하위 노드의 결과가 상태에 남아 다음 판단에 쓰인다. 재호출을 허용하면 같은 노드를 계속 선택할 수도 있으므로 단계 상한이 필요하다. 선택한 노드 이름이 곧 외부 작업 권한을 뜻하지는 않는다. 실제 자료 조회나 코드 실행은 별도 도구 권한으로 제어한다.

[슈퍼바이저 그래프 예시] 아래 researcher와 coder는 흐름을 보여주는 고정 응답 노드이며 검색·코드 실행을 하지 않는다. langgraph, langchain-ollama, pydantic를 설치하고 구조화 출력을 지원하는 Ollama 모델을 OLLAMA_MODEL에 지정한다. 모델이 스키마를 지원하지 않으면 슈퍼바이저 호출에서 실패할 수 있으므로 ChatOllama 연동 문서의 지원 기능을 확인한다.

import os
from typing import Literal

from langchain_core.messages import AIMessage, HumanMessage
from langchain_ollama import ChatOllama
from langgraph.graph import END, START, MessagesState, StateGraph
from pydantic import BaseModel

class RouteDecision(BaseModel):
    next: Literal["researcher", "coder", "FINISH"]

class AgentState(MessagesState):
    next: Literal["researcher", "coder", "FINISH"]
    steps: int

router_model = ChatOllama(
    model=os.environ["OLLAMA_MODEL"], temperature=0
).with_structured_output(RouteDecision)

def supervisor(state: AgentState) -> dict:
    decision = router_model.invoke([
        ("system", "researcher는 자료를 확인하고 coder는 구현 초안을 쓴다. "
                   "대화 기록을 보고 다음 담당자를 researcher, coder, FINISH "
                   "중 하나로 선택하라. 필요한 정보가 없으면 FINISH를 선택하라."),
        *state["messages"],
    ])
    return {"next": decision.next, "steps": state["steps"] + 1}

def researcher(state: AgentState) -> dict:
    # 흐름 확인용 고정 자료. 실제 서비스의 검색 결과가 아니다.
    return {"messages": [AIMessage(
        content="검토할 항목: 입력 길이, 근거가 없는 질문, 출처 ID."
    )]}

def coder(state: AgentState) -> dict:
    # 코드 실행 없이 구현 방향만 반환한다.
    return {"messages": [AIMessage(
        content="초안: 입력 검증 → 허용 문서 검색 → 답변 → 출처 ID 검사."
    )]}

def route(state: AgentState) -> str:
    if state["steps"] >= 4 or state["next"] == "FINISH":
        return "finish"
    return state["next"]

builder = StateGraph(AgentState)
builder.add_node("supervisor", supervisor)
builder.add_node("researcher", researcher)
builder.add_node("coder", coder)
builder.add_edge(START, "supervisor")
builder.add_conditional_edges("supervisor", route, {
    "researcher": "researcher",
    "coder": "coder",
    "finish": END,
})
builder.add_edge("researcher", "supervisor")
builder.add_edge("coder", "supervisor")
graph = builder.compile()

result = graph.invoke({
    "messages": [HumanMessage(content="FAQ 답변 코드를 설계해줘.")],
    "next": "FINISH",
    "steps": 0,
})
print(result["next"], result["steps"])

supervisor는 Pydantic 객체 자체가 아니라 {"next": ..., "steps": ...} 형태의 상태 갱신 딕셔너리를 반환한다. route는 FINISH와 상한을 모두 종료 경로로 보낸다. 하위 노드가 메시지를 남기면 MessagesState의 메시지 병합 규칙에 따라 다음 슈퍼바이저 호출에 포함된다. 이 예제에서는 어떤 순서로 노드가 호출될지 모델 결과에 달렸으므로 특정 출력값을 정답처럼 제시하지 않는다.

두 패턴을 선택하는 기준

성찰은 한 결과물의 품질을 반복해서 개선할 때, 서브그래프는 반복해서 쓸 작업 흐름과 상태 경계가 있을 때, 슈퍼바이저는 다음 전문 작업이 입력에 따라 달라질 때 고려한다. 작업 순서가 이미 정해져 있다면 고정 그래프가 단순하다. 여러 모델을 동원하는 비용과 지연은 각 역할의 이득과 함께 평가해야 한다.

문제	관찰할 실패	조치
같은 비평을 반복	초안이 바뀌지 않음	차이를 검사하고 반복 중단
서브그래프 결과 누락	부모·자식 상태 키 불일치	경계에서 키 변환 테스트
슈퍼바이저 무한 위임	같은 노드를 계속 선택	최대 단계·시간 제한
하위 에이전트의 근거 없는 결과	출처 ID 없음	결과 스키마·근거 검증
권한이 다른 도구를 물려받음	필요 없는 도구 노출	하위 노드별 실행 권한 제한

위의 성찰 출력은 문장이 다듬어졌는지만으로 성공이라 판단할 수 없다. 요구한 세 단락이 맞는지, 원래 질문에 답하는지, 새로운 허위 사실이 생기지 않았는지를 별도 사례로 확인한다. 하위 에이전트가 반환한 결과도 슈퍼바이저가 받았다는 사실만으로 검증되지 않는다. LangGraph 서브그래프 문서와 워크플로 문서를 기준으로 상태 경계와 종료 경로를 설계한다.