에이전트란
에이전트의 구성
에이전트는 다음 요소로 구성된다.
-
모델: 현재 상태를 해석하고 다음 행동을 결정
-
환경: 에이전트가 정보를 읽거나 변경하는 대상
-
도구: 환경을 조회하거나 변경하는 수단
-
피드백: 도구 실행 결과와 변경된 환경 상태
사용자 목표
→ 모델이 행동 결정
→ 도구 실행
→ 결과 반환
→ 모델이 다음 행동 결정
모델과 에이전트의 관계
LLM은 에이전트의 의사결정 구성요소다.
에이전트
= 모델
+ 환경
+ 도구
+ 실행 결과
+ 반복 제어
환경과 도구
| 에이전트 종류 | 환경 | 도구 |
|-- | |- |
| 코딩 에이전트 | 파일시스템, 저장소, 터미널 | 파일 읽기, 코드 수정, 명령 실행 |
| 리서치 에이전트 | 웹페이지, 검색 결과 | 웹 검색, 페이지 조회 |
| 데이터 에이전트 | 데이터베이스 | SQL 조회, 데이터 수정 |
도구는 크게 두 종류로 나뉜다.
-
조회 도구: 파일 읽기, 검색, SQL 조회
-
변경 도구: 파일 수정, 이메일 전송, 코드 실행
워크플로
워크플로에서는 실행 경로를 개발자가 미리 정한다.
입력
→ 검색
→ 요약
→ 저장
→ 종료
LLM이 여러 단계에서 사용되더라도, 실행 순서와 분기를 코드가 결정하면 워크플로다.
documents = retrieve(query)
summary = summarize(documents)
result = save(summary)
에이전트
에이전트에서는 모델이 실행 중 다음 행동을 선택한다.
현재 상태
→ 다음 도구 선택
→ 도구 실행
→ 결과 확인
→ 다음 행동 또는 종료 결정
while True:
response = call_model(messages, tools)
if not response.tool_calls:
return response
results = execute_tools(response.tool_calls)
messages.extend(results)
워크플로와 에이전트의 차이
| 구분 | 워크플로 | 에이전트 |
|---|---|---|
| 실행 경로 | 개발자가 정의 | 모델이 실행 중 선택 |
| 도구 선택 | 코드가 결정 | 모델이 결정 |
| 반복 여부 | 코드 조건으로 결정 | 모델이 결과를 보고 판단 |
| 종료 시점 | 미리 정의 | 실행 중 판단 |
| 경로 예측 | 쉬움 | 실행마다 달라질 수 있음 |
자율성의 정도
워크플로와 에이전트의 차이는 자율성의 정도에 따라 연속적으로 나타난다.
고정 워크플로
→ 일부 단계만 모델이 선택
→ 모델이 도구와 순서를 선택
→ 모델이 작업 분해와 종료까지 판단
자율성은 다음 항목별로 조절할 수 있다.
-
작업 순서
-
도구 선택
-
도구 인자 생성
-
반복 여부
-
종료 조건
-
사용자 승인 여부
정리
워크플로
= 개발자가 실행 경로를 결정
에이전트
= 모델이 도구 결과를 바탕으로 다음 행동을 결정
구분 기준
= 실행 경로를 개발자가 정하는가,
모델이 실행 중 선택하는가
에이전트의 Planning
전체 구조
전체 목표
→ 하위 작업 분해
→ 실행 순서 결정
→ 작업 수행
→ 결과 평가
→ 계획 수정
→ 종료
Planning은 다음 네 가지로 구성된다.
-
작업 분해
-
행동 선택
-
실행 결과 평가
-
재계획
Tool Loop와 Planning
Tool Loop
현재 상태 확인
→ 도구 선택
→ 도구 실행
→ 결과 확인
→ 다음 도구 선택
Planning이 포함된 구조
전체 목표 확인
→ 하위 작업 생성
→ 현재 작업 선택
→ 도구 실행
→ 완료 여부 확인
→ 남은 계획 수정
Tool Loop는 반복 구조이고, Planning은 그 반복에서 무엇을 어떤 순서로 실행할지 관리하는 방식이다.
ReAct
ReAct는 각 반복에서 다음 행동 하나를 결정한다.
현재 상태
→ 다음 행동 결정
→ 도구 실행
→ 결과 관찰
→ 다음 행동 결정
전통적인 표현:
Thought
→ Action
→ Observation
→ Thought
→ Action
→ Observation
예시:
목표: 두 프레임워크 비교
1. 프레임워크 A 검색
2. 검색 결과 확인
3. 프레임워크 B 검색
4. 검색 결과 확인
5. 추가 자료가 필요한지 판단
6. 최종 답변 작성
전체 계획을 미리 고정하지 않고, 실행 결과에 따라 다음 행동을 정한다.
ReAct의 상태
state = {
"goal": "두 프레임워크 비교",
"actions": [],
"observations": [],
}
최소 구조
while True:
action = decide_next_action(
goal=goal,
history=history,
)
if action.type == "finish":
return action.answer
observation = execute_tool(
action.name,
action.arguments,
)
history.append({
"action": action,
"observation": observation,
})
Plan-and-Execute
Plan-and-Execute는 실행 전에 여러 단계의 계획을 만든다.
전체 목표
→ 다단계 계획 생성
→ 각 단계 실행
→ 결과 평가
→ 계획 수정
예시:
목표: 두 프레임워크 비교
1. 프레임워크 A 구조 조사
2. 프레임워크 B 구조 조사
3. 동일 기준으로 기능 비교
4. 구현 코드 확인
5. 최종 결과 작성
작업 간 의존성도 표현할 수 있다.
A 조사 ─┐
├→ 비교 → 코드 검증 → 최종 작성
B 조사 ─┘
A 조사와 B 조사는 병렬 실행할 수 있다.
작업 분해
큰 목표를 실행 가능한 단위로 나눈다.
큰 목표
→ 하위 목표
→ 구체적인 실행 작업
예시:
에이전트 프레임워크 분석
1. 개념 조사
2. 실행 구조 조사
3. 소스코드 확인
4. 차이점 비교
5. 예제 작성
6. 결과 검증
작업 분해 방법:
-
모델이 직접 하위 작업 생성
-
작업 종류별 고정 템플릿 사용
-
사용자가 작업 순서 제공
-
별도 Planner 사용
Planning State
명시적인 계획을 쓰려면 메시지와 별도로 계획 상태를 관리해야 한다.
state = {
"goal": "두 프레임워크 비교",
"plan": [
{
"id": 1,
"task": "프레임워크 A 조사",
"status": "completed",
},
{
"id": 2,
"task": "프레임워크 B 조사",
"status": "running",
},
{
"id": 3,
"task": "기능 비교",
"status": "pending",
},
],
"current_step": 2,
"results": {
"1": "A 조사 결과"
},
}
관리해야 하는 정보:
-
전체 목표
-
전체 작업 목록
-
현재 작업
-
완료된 작업
-
작업별 결과
-
실패한 작업
-
남은 작업
계획 평가
계획을 실행하기 전과 실행 도중에는 다음 항목을 점검한다.
-
목표와 계획이 일치하는가
-
필요한 작업이 빠졌는가
-
실행할 수 없는 작업이 있는가
-
작업 순서가 맞는가
-
병렬 실행 가능한 작업이 있는가
-
현재 결과로 목표가 달성됐는가
구조:
계획 생성
→ 계획 검토
→ 실행
→ 결과 검토
Replanning
실행 결과에 따라 계획을 수정한다.
계획
→ 실행
→ 결과 확인
→ 누락 또는 실패 발견
→ 계획 수정
예시:
초기 계획
1. 평가 방법 조사
2. 평가 지표 조사
3. 최종 정리
실행 결과:
평가자 간 불일치 처리 방법이 빠짐
수정된 계획:
1. 평가 방법 조사 완료
2. 평가 지표 조사 완료
3. 평가자 불일치 측정 방법 조사
4. 평가 결과 보정 방법 조사
5. 최종 정리
Reflection
실행 결과를 평가해 이후의 행동과 계획 수정에 반영한다.
확인할 내용:
-
어떤 작업이 완료됐는가
-
어떤 작업이 실패했는가
-
부족한 정보가 무엇인가
-
같은 행동을 반복하고 있는가
-
기존 계획을 계속 사용해도 되는가
reflection = {
"completed": [
"기본 평가 방법 조사"
],
"missing": [
"평가자 불일치 처리"
],
"failed": [],
}
실행 결과
→ 완료·누락·실패 구분
→ 다음 행동 결정
→ 필요하면 계획 수정
ReAct와 Plan-and-Execute 비교
| 구분 | ReAct | Plan-and-Execute |
|---|---|---|
| 계획 범위 | 다음 행동 하나 | 여러 단계 |
| 계획 생성 | 매 반복마다 | 실행 전 |
| 주요 상태 | 행동과 관찰 이력 | 계획과 작업 상태 |
| 환경 변화 대응 | 즉시 다음 행동 변경 | 계획 수정 |
| 병렬 실행 | 기본적으로 어려움 | 가능 |
| 진행률 관리 | 이력에서 추론 | 상태로 명시 |
| 작업 의존성 | 문맥에서 판단 | 계획에 명시 |
| 적합한 작업 | 탐색적 작업 | 구조화된 복합 작업 |
혼합 구조
두 방식을 함께 조합할 수도 있다.
전체 목표
→ Plan-and-Execute로 하위 작업 생성
→ 각 하위 작업은 ReAct로 실행
→ 결과 평가
→ 전체 계획 수정
예시:
전체 계획
1. 자료 조사
2. 소스코드 분석
3. 비교
4. 글 작성
각 단계 내부:
현재 상태 확인
→ 검색
→ 결과 확인
→ 추가 검색
→ 단계 완료
한계
ReAct
-
도구 호출이 순차적으로 늘어남
-
전체 진행 상황이 명시되지 않을 수 있음
-
같은 행동을 반복할 수 있음
-
최대 반복 횟수 같은 제한 필요
Plan-and-Execute
-
초기 계획이 틀리면 여러 작업이 잘못될 수 있음
-
환경이 자주 바뀌면 계속 재계획해야 함
-
탐색적인 문제는 사전 계획이 어려움
-
계획 생성과 평가에 추가 모델 호출이 필요함
정리
ReAct
= 다음 행동 하나를 정하고 실행 결과를 본 뒤 다시 판단
Plan-and-Execute
= 전체 목표를 여러 작업으로 나누고 순서대로 실행
Task Decomposition
= 큰 목표를 실행 가능한 작업으로 분해
Reflection
= 실행 결과에서 완료·누락·실패를 확인
Replanning
= 평가 결과를 반영해 기존 계획을 수정
Planning
= 작업 분해
+ 실행 순서 관리
+ 결과 평가
+ 계획 수정
create_agent
create_agent는 다음 구조를 LangGraph로 만든다.
상태
→ 모델 호출
→ 도구 호출 여부 판단
→ 도구 실행
→ 결과를 상태에 추가
→ 모델 재호출
create_agent는 모델 응답에 tool_calls가 있으면 해당 도구를 실행하고, 결과를 ToolMessage로 상태에 추가한다. tool_calls가 없을 때까지 이 순서를 반복한다.
상태
에이전트는 이전 행동과 실행 결과를 상태에 저장한다.
class AgentState(TypedDict):
messages: list[AnyMessage]
상태에는 메시지가 순차적으로 누적된다.
HumanMessage
→ AIMessage(tool_calls)
→ ToolMessage
→ AIMessage(tool_calls)
→ ToolMessage
→ AIMessage(final answer)
모델은 이 상태를 읽고 다음 행동을 결정한다.
state = {
"messages": [
HumanMessage(content="두 프레임워크를 비교해줘"),
AIMessage(
content="",
tool_calls=[
{
"name": "search",
"args": {"query": "framework A"},
}
],
),
ToolMessage(
content="검색 결과",
tool_call_id="call_1",
),
]
}
모델 노드
모델 노드는 현재 상태와 사용할 수 있는 도구를 입력받는다.
def model_node(state):
model_with_tools = model.bind_tools(tools)
response = model_with_tools.invoke(
state["messages"]
)
return {
"messages": [response]
}
모델 출력은 두 가지다.
AIMessage(tool_calls=[...])
→ 행동 계속
AIMessage(content="최종 답변")
→ 종료
create_agent는 ModelRequest에 모델, 도구, 시스템 메시지, 현재 메시지와 상태를 담아 모델을 실행한다.
관련 흐름을 간략화한 코드는 아래와 같다.
request = ModelRequest(
model=model,
tools=tools,
messages=state["messages"],
state=state,
)
response = execute_model(request)
모델은 함수를 직접 실행하는 대신 도구 이름과 인자를 생성한다.
모델
→ 도구 이름과 인자 생성
에이전트 런타임
→ 해당 함수를 찾아 실행
도구 노드
도구 노드는 모델이 생성한 tool_calls를 실행한다.
def tool_node(state):
last_message = state["messages"][-1]
results = []
for tool_call in last_message.tool_calls:
tool = tools_by_name[tool_call["name"]]
result = tool.invoke(
tool_call["args"]
)
results.append(
ToolMessage(
content=result,
tool_call_id=tool_call["id"],
)
)
return {
"messages": results
}
LangGraph에서는 ToolNode가 이 역할을 담당한다.
tool_node = ToolNode(tools)
AIMessage.tool_calls
→ ToolNode
→ 함수 실행
→ ToolMessage
라우팅
모델 실행 뒤에는 라우터가 다음 노드를 선택한다.
def route_after_model(state):
last_message = state["messages"][-1]
if last_message.tool_calls:
return "tools"
return END
도구 실행 뒤에는 제어가 다시 모델 노드로 돌아간다.
graph.add_edge(
"tools",
"model",
)
전체 그래프:
graph = StateGraph(AgentState)
graph.add_node(
"model",
model_node,
)
graph.add_node(
"tools",
ToolNode(tools),
)
graph.add_edge(
START,
"model",
)
graph.add_conditional_edges(
"model",
route_after_model,
{
"tools": "tools",
END: END,
},
)
graph.add_edge(
"tools",
"model",
)
agent = graph.compile()
START
↓
model
├─ tool_calls 있음 → tools ─┐
│ │
└─ tool_calls 없음 → END │
│
model ◀────────┘
create_agent는 StateGraph에 model과 tools 노드를 추가하고, 두 노드 사이에 조건부 엣지를 만든다.
while 루프와 그래프
같은 루프를 일반 코드로 작성하면 아래와 같다.
while True:
response = call_model(state)
state["messages"].append(response)
if not response.tool_calls:
return response
tool_results = execute_tools(
response.tool_calls
)
state["messages"].extend(
tool_results
)
그래프 구현은 이 루프를 다음 요소로 분리한다.
상태
= messages
판단
= model node
행동
= tools node
분기
= conditional edge
반복
= tools → model edge
이것이 create_agent의 기본 흐름이다.
미들웨어로 루프 확장하기
모델 노드에 모든 기능을 넣으면 책임이 과도하게 커진다.
def model_node(state):
check_limits(state)
summarize_messages(state)
select_tools(state)
log_request(state)
response = model.invoke(...)
validate_response(response)
log_response(response)
return response
미들웨어는 제한 확인, 로깅, 응답 검증 같은 공통 처리를 모델 호출과 분리한다.
class CustomMiddleware(AgentMiddleware):
def before_model(
self,
state,
runtime,
):
return update_state(state)
def wrap_model_call(
self,
request,
handler,
):
request = modify_request(request)
response = handler(request)
return modify_response(response)
create_agent는 미들웨어 훅을 그래프 노드로 등록한다.
START
→ before_model
→ model
→ after_model
→ tools
→ before_model
미들웨어를 그래프에 등록하는 과정을 간략화하면 아래와 같다.
graph.add_node(
"model",
model_node,
)
for middleware in middleware_list:
if middleware.has_before_model:
graph.add_node(
f"{middleware.name}.before_model",
middleware.before_model,
)
if middleware.has_after_model:
graph.add_node(
f"{middleware.name}.after_model",
middleware.after_model,
)
미들웨어는 기본 루프의 실행 전후에 필요한 로직을 삽입한다.
기본 루프
model → tools → model
미들웨어
= 기본 루프의 실행 전후에 로직 삽입
create_deep_agent
create_deep_agent는 기능별 미들웨어를 구성해 create_agent에 전달한다.
middleware = [
TodoListMiddleware(),
FilesystemMiddleware(
backend=backend,
),
SubAgentMiddleware(
backend=backend,
subagents=subagents,
),
create_summarization_middleware(
model,
backend,
),
]
return create_agent(
model=model,
tools=tools,
middleware=middleware,
)
create_deep_agent
→ 기능별 미들웨어 구성
→ create_agent 호출
→ 동일한 Tool Loop 실행
계획을 상태로 분리
기본 에이전트에서는 계획을 메시지 안에만 저장한다.
AIMessage:
먼저 A를 조사하고,
다음에 B를 조사한 뒤,
두 결과를 비교해야겠다.
Todo 기반 구조에서는 계획을 별도 상태로 관리한다.
class PlanningState(AgentState):
todos: list[Todo]
todos = [
{
"content": "A 조사",
"status": "completed",
},
{
"content": "B 조사",
"status": "in_progress",
},
{
"content": "결과 비교",
"status": "pending",
},
]
write_todos 도구는 Todo 목록과 메시지를 한 번에 갱신한다.
def write_todos(
todos,
tool_call_id,
):
return Command(
update={
"todos": todos,
"messages": [
ToolMessage(
content="Todo updated",
tool_call_id=tool_call_id,
)
],
}
)
TodoListMiddleware는 todos 상태와 write_todos 도구를 추가한다.
암묵적인 계획
→ messages 안에서 유지
명시적인 계획
→ todos 상태로 분리
작업 결과를 파일로 분리
도구 결과를 모두 메시지에 넣으면 작업이 진행될수록 상태가 커진다.
state["messages"].append(
ToolMessage(
content=very_large_search_result
)
)
작업 공간을 사용하면 결과를 파일에 분리해 저장할 수 있다.
path = "/research/framework_a.md"
backend.write(
path,
search_result,
)
return ToolMessage(
content=f"Saved to {path}"
)
messages
= 현재 판단에 필요한 정보
filesystem
= 조사 결과와 중간 산출물
에이전트는 후속 작업에 필요할 때 파일을 다시 읽는다.
content = backend.read(
"/research/framework_a.md"
)
FilesystemMiddleware는 이 구조를 read_file, write_file, edit_file, glob, grep 같은 도구로 제공한다.
오래된 실행 기록 압축
Tool Loop가 반복되면 메시지가 계속 증가한다.
사용자 요청
모델 응답
도구 결과
모델 응답
도구 결과
모델 응답
도구 결과
...
메시지가 일정 크기를 넘으면 오래된 기록을 요약한다.
def summarize_history(state):
old_messages = state["messages"][:-10]
recent_messages = state["messages"][-10:]
summary = summarizer.invoke(
old_messages
)
return {
"messages": [
SystemMessage(
content=f"Previous work: {summary}"
),
*recent_messages,
]
}
오래된 상세 실행 기록
→ 요약
최근 실행 기록
→ 그대로 유지
중간 산출물
→ 파일에 저장
create_deep_agent는 오래된 기록을 요약하는 미들웨어를 기본 스택에 포함한다.
작업을 다른 에이전트에 위임
서브에이전트도 자체 도구와 실행 루프를 가진 독립적인 에이전트다.
researcher = create_agent(
model=research_model,
tools=[
search,
read_page,
],
)
메인 에이전트는 task 도구를 통해 서브에이전트를 호출한다.
def task(
agent_name,
description,
):
agent = agents[agent_name]
result = agent.invoke({
"messages": [
HumanMessage(
content=description
)
]
})
return result["messages"][-1].content
Main Agent
→ task 도구 호출
→ Subagent의 자체 Tool Loop
→ 최종 결과 반환
→ Main Agent 계속 실행
Deep Agents의 서브에이전트 생성 코드는 각 서브에이전트의 설정을 create_agent에 전달한다.
def create_sub_agent(spec):
return create_agent(
model=spec["model"],
tools=spec["tools"],
system_prompt=spec["system_prompt"],
middleware=spec["middleware"],
)
멀티에이전트 구조의 실행 흐름:
에이전트 하나
= 하나의 Tool Loop
멀티에이전트
= 다른 Tool Loop를 호출할 수 있는 Tool Loop