인공지능은 이제 단일 모델 기반의 응답 시스템에서 벗어나, 자율적인 구성 요소들이 추론하고 행동하며 협력하는 에이전트 중심 시스템으로 급격히 이동 중이다. 기존 LLM의 단발성 상호작용과 달리, 에이전트 아키텍처는 컨텍스트가 유지되는 상태 기반 세션, 작업별 특화 로직의 모듈화, 외부 도구 및 워크플로우와의 상호운용성을 지향한다. 이 글은 레퍼런스를 참고해 FastAPI, LangGraph, MCP(Model Context Protocol)를 통해 확장 가능한 플랫폼 구조를 정리한다.
프로젝트 아키텍처 및 주요 구성 요소
시스템은 클라이언트-API-오케스트레이션-도구 실행으로 이어지는 명확한 계층형 구조를 가진다. FastAPI는 고성능 비동기 웹 서비스를 통해 에이전트의 진입점 역할을 수행하며, 실제 추론 로직과 분리되어 있어 유지보수가 용이하다. 서비스 레이어는 API와 오케스트레이션 계층을 연결하며 세션 초기화와 최종 응답 포맷팅을 담당한다. LLM 제공자 추상화를 통해 OpenAI나 Anthropic과 같은 다양한 모델을 유연하게 선택할 수 있는 구조이다.
LangGraph를 이용한 에이전트 오케스트레이션
에이전트 아키텍처의 핵심은 LangGraph를 이용한 상태 관리와 워크플로우 정의이다. StateGraph를 활용하여 에이전트가 언제 도구를 호출하고 언제 작업을 종료할지 명시적으로 제어한다. 이는 에이전트의 행동을 투명하게 만들고 디버깅을 용이하게 하는 핵심적인 설계이다.
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 패턴 구현 예시
랭그래프로 리액트 패턴을 구현해 본다. 이 에이전트는 사용자 질문에 대한 특정 도구를 발견하지 못할때까지 반복해 호출하도록 구현되었다.
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)
AI 에이전트 패턴
앞서 본 것처럼 에이전트 개발에는 다양한 패턴이 있을 수 있다. 이는 유스케이스 목적에 따라 결정되어야 한다. 토큰 사용량 대비 효과적인 답을 낼 수 있도록 구현되어야 한다. 다음은 다양한 에이전트 패턴 구현 방법을 보여준다.
MCP 기반의 도구 통합 및 확장성
MCP(Model Context Protocol)는 모델과 도구 간의 상호작용을 표준화하여 시스템의 신뢰성을 높인다. 모든 도구는 @tool 데코레이터를 사용하여 선언적으로 정의되며, 중앙 집중식 레지스트리 패턴을 통해 관리된다. 이러한 방식은 로컬 도구를 외부 MCP 서버로 교체하거나 런타임에 새로운 기능을 동적으로 로드하는 것을 매우 간편하게 만든다. 결과적으로 에이전트는 단순한 텍스트 생성을 넘어 외부 API 호출, 계산, 데이터 분석 등 실질적인 액션을 수행하는 능력을 갖추게 된다.
결론
이 예시는 Docker 및 Docker Compose를 지원하여 일관된 런타임 환경과 클라우드 배포의 편의성을 확보하였다. 다중 에이전트 간의 협력 체인 구축, 세션 내 인간 피드백 통합, 워크플로우 관찰 가능성(Observability) 강화 등을 통해 더욱 고도화된 에이전트 시스템으로 확장이 가능하다.
댓글 없음:
댓글 쓰기