인공지능은 이제 단일 모델 기반의 응답 시스템에서 벗어나, 자율적인 구성 요소들이 추론하고 행동하며 협력하는 에이전트 중심 시스템으로 급격히 이동 중이다. 기존 LLM의 단발성 상호작용과 달리, 에이전트 아키텍처는 컨텍스트가 유지되는 상태 기반 세션, 작업별 특화 로직의 모듈화, 외부 도구 및 워크플로우와의 상호운용성을 지향한다. FastAPI, LangGraph, MCP(Model Context Protocol)를 통해 확장 가능한 플랫폼 구조와 디자인 패턴을 정리한다. 좀 더 상세한 내용은 레퍼런스를 참고한다.
프로젝트 아키텍처 및 주요 구성 요소
시스템은 클라이언트-API-오케스트레이션-도구 실행으로 이어지는 명확한 계층형 구조를 가진다. FastAPI는 고성능 비동기 웹 서비스를 통해 에이전트의 진입점 역할을 수행하며, 실제 추론 로직과 분리되어 있어 유지보수가 용이하다. 서비스 레이어는 API와 오케스트레이션 계층을 연결하며 세션 초기화와 최종 응답 포맷팅을 담당한다. LLM 제공자 추상화를 통해 OpenAI나 Anthropic과 같은 다양한 모델을 유연하게 선택할 수 있는 구조이다.
MCP(Model Context Protocol)는 모델과 도구 간의 상호작용을 표준화하여 시스템의 신뢰성을 높인다. 모든 도구는 @tool 데코레이터를 사용하여 선언적으로 정의되며, 중앙 집중식 레지스트리 패턴을 통해 관리된다. 이러한 방식은 로컬 도구를 외부 MCP 서버로 교체하거나 런타임에 새로운 기능을 동적으로 로드하는 것을 매우 간편하게 만든다. 결과적으로 에이전트는 단순한 텍스트 생성을 넘어 외부 API 호출, 계산, 데이터 분석 등 실질적인 액션을 수행하는 능력을 갖추게 된다.
이제 다이어그램에서 전체 에이전트 패턴 구조와 몇몇 핵심적인 부분을 구현해 본다.
LangGraph를 이용한 에이전트 오케스트레이션
에이전트 아키텍처의 핵심은 LangGraph를 이용한 상태 관리와 워크플로우 정의이다. StateGraph를 활용하여 에이전트가 언제 도구를 호출하고 언제 작업을 종료할지 명시적으로 제어한다. 이는 에이전트의 행동을 투명하게 만들고 디버깅을 용이하게 하는 핵심적인 설계이다.
다음과 같은 에이전트 구조가 있다고 치자. agent는 사용자 입력을 받아 LLM 추론을 한다. 그 결과 tools 호출(예. 날씨, 온도, IoT 센서, 데이터베이스, 로보틱스 액추에이터 등등)이 필요하면 tools를 호출하고, 그 결과를 다시 LLM에게 전달해 답변을 출력한다. 아니면, END 종료한다.
이를 랭그래프로 구현하면 다음과 같다.
from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import ToolNode
def create_agent_graph():
llm = get_llm()
tools = get_tools()
llm_with_tools = llm.bind_tools(tools)
workflow = StateGraph(AgentState)
# LLM 추론 노드 정의
async def call_model(state: AgentState) -> dict:
response = await llm_with_tools.ainvoke(state["messages"])
return {"messages": [response]}
# 노드 등록 및 경로 설정
workflow.add_node("agent", call_model)
workflow.add_node("tools", ToolNode(tools))
workflow.add_edge(START, "agent")
# 조건부 엣지: 도구 호출 여부에 따라 경로 결정
def should_continue(state: AgentState) -> str:
last_message = state["messages"][-1]
if hasattr(last_message, "tool_calls") and last_message.tool_calls:
return "tools"
return END
workflow.add_conditional_edges("agent", should_continue, ["tools", END])
workflow.add_edge("tools", "agent")
return workflow.compile()
LangGraph 기반 ReAct 패턴 구현
사용자 질문에 대해 계속 도구를 호출해 반복된 구조로 답을 찾는 에이전트는 ReAct 구조를 사용할 수 있다. 앞의 구조에서 좀 더 실용적인 도구를 사용해 보겠다. 위키피디아 검색 및 계산기 도구를 호출하도록 랭그래프를 다음과 같이 수정한다.
이 에이전트는 사용자 질문에 대한 특정 도구를 발견하지 못할때까지 반복해 호출할 것이다. 실제 예에서는 토큰 비용 및 성능도 고려해야 하므로, 그래프 LLM 호출 최적화가 필요할 것이다.
from typing import Annotated, Literal
from typing_extensions import TypedDict
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage
from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import ToolNode
from dotenv import load_dotenv
load_dotenv()
# 1. 도구(Tool) 정의
from langchain_core.tools import tool
@tool
def search_wikipedia(query: str) -> str:
"""Search Wikipedia for information. Use this when you need factual information."""
# 실제로는 Wikipedia API 호출
return f"Wikipedia search results for '{query}': [모의 검색 결과 - 실제로는 API 호출]"
@tool
def calculator(expression: str) -> str:
"""Calculate mathematical expressions. Input should be a valid Python expression."""
try:
result = eval(expression, {"__builtins__": {}})
return str(result)
except Exception as e:
return f"Error: {e}"
tools = [search_wikipedia, calculator]
tool_node = ToolNode(tools)
# 2. State 정의
class AgentState(TypedDict):
messages: Annotated[list, "The messages in the conversation"]
# 3. LLM 설정 (tool binding)
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
llm_with_tools = llm.bind_tools(tools)
# 4. Agent 노드 - LLM이 판단
def agent_node(state: AgentState) -> dict:
"""LLM이 도구를 사용할지, 최종 답변을 할지 결정 (Thought)"""
response = llm_with_tools.invoke(state["messages"])
return {"messages": [response]}
# 5. 라우팅 로직 - 종료 조건 결정
def should_continue(state: AgentState) -> Literal["tools", "end"]:
"""
ReAct의 핵심 종료 조건:
- LLM이 tool_calls를 반환하면 → "tools" (Action 필요)
- tool_calls가 없으면 → "end" (최종 답변 완성)
"""
last_message = state["messages"][-1]
if hasattr(last_message, "tool_calls") and last_message.tool_calls:
return "tools" # 도구 사용 필요
return "end" # 최종 답변 완성
# 6. 그래프 구성
workflow = StateGraph(AgentState)
workflow.add_node("agent", agent_node) # Thought 단계
workflow.add_node("tools", tool_node) # Action 단계
workflow.add_edge(START, "agent")
# ReAct 사이클: agent → tools → agent → tools → ... → end
workflow.add_conditional_edges(
"agent",
should_continue, # ← 종료 조건 검출
{
"tools": "tools", # 도구 실행 후 다시 agent로
"end": END # 답변 완성 시 종료
}
)
workflow.add_edge("tools", "agent") # Observation 후 다시 Thought
react_agent = workflow.compile()
# 7. 실행 예시
if __name__ == "__main__":
# 예시 1: 계산 필요
result = react_agent.invoke({
"messages": [HumanMessage(content="What is 342 * 67?")]
})
print("답변:", result["messages"][-1].content)
print("\n" + "="*80 + "\n")
# 예시 2: 검색 + 계산 필요
result = react_agent.invoke({
"messages": [HumanMessage(
content="Search for the population of Seoul, then multiply it by 2"
)]
})
print("답변:", result["messages"][-1].content)
이런 방식으로 필요한 도구들, 목적별 LLM 에이전트(예. 조사, 설계, 아이디어 구현, 요약, 보고서 작성 등)를 추가해 나갈 수 있다.
결론
이 간단한 에이전트 아키턱쳐 예시는 다중 에이전트 간의 협력 체인 구축, 세션 내 인간 피드백 통합, 워크플로우 관찰 가능성(Observability) 강화 등을 통해 더욱 고도화된 에이전트 시스템으로 확장이 가능하다. 본 글에서는 랭그래프를 사용했다. 요즘에는 랭그래프 라이브러리 안전성이 높아져 이런 멀티 에이전트 구조는 큰 문제 없이 개발이 가능해 졌다. 멀티에이전트 개발 개념은 모두 유사하므로 유스케이스 목적에 따라 다른 프레임웍 라이브러리 사용하는 것도 고려할 수 있을 것이다.
부록: AI 에이전트 디자인 패턴
앞서 본 것처럼 에이전트 개발에는 여러 디자인 패턴이 있을 수 있다. 이는 유스케이스 목적에 따라 결정되어야 한다. 토큰 사용량 대비 효과적인 답을 낼 수 있도록 구현되어야 한다. 다음은 다양한 에이전트 패턴 구현 방법을 보여준다.
import operator
from typing import Annotated, List, TypedDict, Literal
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from dotenv import load_dotenv
load_dotenv()
# 순차적 에이전트(Sequential Agent)는 여러 단계에 걸쳐 내부 추론을 수행. 스크래치패드(scratchpad)라는 방식으로 진행. 별도의 외부 도구를 전혀 사용하지 않음.
# 순수한 분석과 연역적 추론만으로도 충분한 상황에 주로 사용.
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
class State(TypedDict):
question: str
steps: Annotated[List[str], operator.add]
answer: str
def plan_node(state: State) -> dict:
sys = (
"당신은 신중한 계획자입니다. 사용자의 질문을 2-4개의 간결한 단계로 나누세요. "
"문제를 해결하지는 마세요. 번호가 매겨진 단계 목록만 반환하고, 추가 텍스트는 작성하지 마세요."
)
messages = [("system", sys), ("user", state["question"])]
resp = llm.invoke(messages)
raw = resp.content
steps = []
for line in str(raw).splitlines():
line = line.strip()
if not line:
continue
line = line.lstrip("-• ").split(". ", 1)[-1] if ". " in line[:4] else line.lstrip("-• ")
steps.append(line)
return {"steps": steps}
def solve_node(state: State) -> dict:
"""계획된 단계를 사용하여 최종 답변만 도출합니다."""
sys = (
"제공된 단계를 사용하여 문제를 해결하세요. "
"최종 답변만 반환하고, 추론 과정은 포함하지 마세요."
)
messages = [
("system", sys),
("user", f"질문: {state['question']}\\\\n단계: {state['steps']}"),
]
resp = llm.invoke(messages)
return {"answer": str(resp.content).strip()}
# Wire up the graph
graph = StateGraph(State)
graph.add_node("plan", plan_node)
graph.add_node("solve", solve_node)
graph.add_edge(START, "plan")
graph.add_edge("plan", "solve")
graph.add_edge("solve", END)
cot_graph = graph.compile()
state = {
"question": "강의 동영상이 120개이고 하루에 15개를 본다면, 완강하는 데 며칠이 걸릴까요?",
"steps": [],
"answer": ""
}
out = cot_graph.invoke(state)
print("최종 답변:", out["answer"])
# 커스텀 에이전트(Custom Agent)는 유연성을 제공. 사용자는 전체적인 로직과 라우팅을 직접 설계할 수 있음. 또한 시스템을 구성하는 개별 노드까지 스스로 정의. 사용자의 요구에 맞춰 자유롭게 맞춤형 제어가 가능.
class CustomState(TypedDict):
input: str
task: Literal["math", "capitalize", "count"]
result: str
def route(state: CustomState) -> str:
"""Deterministic router based on a simple protocol in the input."""
text = state["input"].strip().lower()
if text.startswith("math:"):
return "math"
if text.startswith("capitalize:"):
return "capitalize"
if text.startswith("count:"):
return "count"
return "count"
def do_math(state: CustomState) -> dict:
expr = state["input"].split(":", 1)[-1].strip()
allowed = set("0123456789+-*/(). ")
if any(c not in allowed for c in expr):
return {"result": "Error: unsupported characters in math expression."}
try:
res = eval(expr, {"__builtins__": {}})
except Exception as e:
res = f"Error: {e}"
return {"result": str(res)}
def do_capitalize(state: CustomState) -> dict:
text = state["input"].split(":", 1)[-1].strip()
return {"result": text.upper()}
def do_count(state: CustomState) -> dict:
text = state["input"].split(":", 1)[-1].strip()
tokens = [t for t in text.split() if t]
return {"result": f"words={len(tokens)} chars={len(text)}"}
graph = StateGraph(CustomState)
graph.add_node("math", do_math)
graph.add_node("capitalize", do_capitalize)
graph.add_node("count", do_count)
graph.add_conditional_edges(
START,
route,
{
"math": "math",
"capitalize": "capitalize",
"count": "count",
},
)
graph.add_edge("math", END)
graph.add_edge("capitalize", END)
graph.add_edge("count", END)
custom_agent = graph.compile(debug=True)
for user_input in [
"math: (16 + 3) * 2 + 5",
"capitalize: hello world from AI agent",
"count: 여기에 몇 개의 단어가 있나요?",
]:
out = custom_agent.invoke({"input": user_input, "task": "count", "result": ""})
print(f"입력: {user_input}\\n결과: {out['result']}\\n---")
# 슈퍼바이저(Supervisor) 패턴은 중앙 제어형 에이전트가 전체 작업을 관리합. 상위 에이전트가 문제를 분석한 뒤 이를 여러 하위 노드에 나누어 배정. 각 하위 노드가 작업을 마치면 그 결과를 다시 취합하고 검토.복잡한 작업을 나누어 병렬로 처리하거나 중앙 통제가 필요할 때 유용.
class SupervisorState(TypedDict):
"""여러 에이전트가 있는 슈퍼바이저 패턴의 상태."""
topic: str
messages: Annotated[List[str], operator.add]
next_agent: str
final_answer: str
def researcher_agent(state: SupervisorState) -> dict:
"""연구자 에이전트는 주제에 대한 정보를 수집합니다."""
sys = (
"당신은 연구자입니다. 주어진 주제에 대한 핵심 사실과 정보를 "
"수집하는 것이 당신의 임무입니다. 2-3개의 핵심 포인트를 제공하세요. 간결하게 작성하세요."
)
messages_for_llm = [
("system", sys),
("user", f"다음 주제를 조사하세요: {state['topic']}")
]
resp = llm.invoke(messages_for_llm)
research_msg = f"연구자: {resp.content}"
return {"messages": [research_msg]}
def expert_agent(state: SupervisorState) -> dict:
"""전문가 에이전트는 연구를 기반으로 분석하고 통찰력을 제공합니다."""
sys = (
"당신은 전문 분석가입니다. 제공된 연구를 검토하고 "
"전문가 분석과 결론을 제공하세요. 구체적이고 통찰력 있게 작성하세요."
)
# 이전 메시지에서 컨텍스트 가져오기
context = "\n".join(state["messages"])
messages_for_llm = [
("system", sys),
("user", f"주제: {state['topic']}\n\n이전 조사 내용:\n{context}\n\n전문가 분석을 제공하세요.")
]
resp = llm.invoke(messages_for_llm)
expert_msg = f"전문가: {resp.content}"
return {"messages": [expert_msg]}
def supervisor_agent(state: SupervisorState) -> dict:
"""슈퍼바이저는 다음에 어떤 에이전트가 활동할지 또는 토론을 종료할지 결정합니다."""
sys = (
"당신은 연구자와 전문가 간의 조사 토론을 관리하는 슈퍼바이저입니다. "
"지금까지의 대화를 바탕으로 다음에 무엇을 해야 할지 결정하세요:\n"
"- 초기 조사나 추가 정보가 필요하면 'researcher'를 반환하세요\n"
"- 조사가 완료되고 전문가 분석이 필요하면 'expert'를 반환하세요\n"
"- 조사와 전문가 분석이 모두 완료되면 'end'를 반환하세요\n\n"
"단 하나의 단어만 응답하세요: researcher, expert, 또는 end"
)
context = "\n".join(state["messages"]) if state["messages"] else "아직 토론이 없습니다"
messages_for_llm = [
("system", sys),
("user", f"주제: {state['topic']}\n\n대화 내용:\n{context}\n\n다음은 무엇인가요?")
]
resp = llm.invoke(messages_for_llm)
next_step = resp.content.strip().lower()
# 유효한 응답인지 확인
if next_step not in ["researcher", "expert", "end"]:
next_step = "end"
return {"next_agent": next_step}
def finalize_answer(state: SupervisorState) -> dict:
"""토론에서 최종 답변을 작성합니다."""
sys = (
"조사 토론을 명확하고 간결한 최종 답변으로 요약하세요. "
"핵심 발견 사항과 전문가 통찰력을 포함하세요."
)
context = "\n".join(state["messages"])
messages_for_llm = [
("system", sys),
("user", f"주제: {state['topic']}\n\n토론 내용:\n{context}\n\n최종 요약을 제공하세요:")
]
resp = llm.invoke(messages_for_llm)
return {"final_answer": resp.content}
def route_supervisor(state: SupervisorState) -> str:
"""슈퍼바이저의 결정에 따라 라우팅합니다."""
next_agent = state.get("next_agent", "researcher")
if next_agent == "end":
return "finalize"
return next_agent
supervisor_graph = StateGraph(SupervisorState)
supervisor_graph.add_node("supervisor", supervisor_agent)
supervisor_graph.add_node("researcher", researcher_agent)
supervisor_graph.add_node("expert", expert_agent)
supervisor_graph.add_node("finalize", finalize_answer)
supervisor_graph.add_edge(START, "supervisor")
supervisor_graph.add_conditional_edges(
"supervisor",
route_supervisor,
{
"researcher": "researcher",
"expert": "expert",
"finalize": "finalize"
}
)
supervisor_graph.add_edge("researcher", "supervisor")
supervisor_graph.add_edge("expert", "supervisor")
supervisor_graph.add_edge("finalize", END)
supervisor_agent_graph = supervisor_graph.compile(debug=True)
topic = "AI 에이전트를 구축하는 데 LangGraph를 사용하는 주요 이점은 무엇인가요?"
initial_state = {
"topic": topic,
"messages": [],
"next_agent": "",
"final_answer": ""
}
result = supervisor_agent_graph.invoke(initial_state)
print(f"주제: {topic}\n")
print("\n토론 내용:")
for msg in result["messages"]:
print(f"\n{msg}\n")
print(f"\n최종 답변:\n{result['final_answer']}")
댓글 없음:
댓글 쓰기