[온프레미스 AI Assistant 아키텍처 — Qwen3·Qwen3-VL 14일 설계] 12/14화: Qwen3 Function Calling·MCP·에이전트 오케스트레이션 실전 설계
이 글은 「온프레미스 AI Assistant 아키텍처 — Qwen3·Qwen3-VL 14일 설계」 시리즈 12일차로, Qwen3 Function Calling과 MCP 프로토콜을 활용한 에이전트 오케스트레이션 실전 설계를 다룹니다.
어제 11화에서 단기·장기·세션 메모리 아키텍처를 설계했습니다. 모델이 과거 대화를 기억하고, 사용자 선호를 축적하며, 세션 경계를 넘어 연속성을 유지하는 구조까지 완성했죠. 하지만 기억만 하는 AI는 비서가 아닙니다. 실제로 행동하는 AI—사내 시스템을 조회하고, 파일을 생성하며, 외부 API를 호출하는 에이전트—가 되어야 비로소 온프레미스 AI Assistant의 가치가 실현됩니다.
오늘은 Phase D(에이전트·안전·종합)의 첫 번째 화로, Qwen3의 도구 사용 능력을 실전에서 어떻게 신뢰성 있게 활용하고, MCP(Model Context Protocol)로 사내 시스템과 연동하며, ReAct·Plan-Execute·멀티 에이전트 패턴으로 복잡한 워크플로를 오케스트레이션하는지 다룹니다.
오늘의 핵심 3가지
- Qwen3 Function Calling 신뢰도 확보 — tool_call JSON 파싱 실패율을 1% 미만으로 낮추는 프롬프트·파싱·검증 3중 전략
- MCP 기반 사내 시스템 연동 — 표준 프로토콜로 DB 조회·Jira·슬랙·사내 REST API를 툴로 노출하고, 권한·감사 로그까지 일관 관리
- 에이전트 오케스트레이션 — ReAct(단일 추론-행동 루프), Plan-Execute(계획 후 실행), 멀티 에이전트 라우팅, LangGraph 상태 그래프까지 실전 설계
1. LLM 에이전트의 기본 구조 — 왜 도구가 필요한가
LLM은 본질적으로 텍스트 생성기입니다. 아무리 똑똑해도 실시간 주가를 알 수 없고, 데이터베이스에 레코드를 삽입할 수 없으며, 이메일을 보낼 수도 없습니다. 이 간극을 메우는 것이 도구 사용(Tool Use)이고, 도구를 여러 단계에 걸쳐 조합하는 것이 에이전트(Agent)입니다.
에이전트의 기본 루프는 단순합니다:
- 관찰(Observe) — 사용자 질문 + 컨텍스트(메모리·RAG 결과)를 수집
- 사고(Think) — 어떤 도구를 어떤 인자로 호출할지 결정
- 행동(Act) — 도구를 실행하고 결과를 받아옴
- 응답(Respond) — 도구 결과를 종합해 최종 답변 생성
이 루프가 한 번이면 단순 도구 호출, 여러 번 반복되면 에이전트입니다. Qwen3는 이 루프를 위한 네이티브 Function Calling을 지원하며, 이것이 오늘 이야기의 출발점입니다.
2. Qwen3 Function Calling 구조와 동작 원리
2-1. 도구 정의 형식
Qwen3의 Function Calling은 OpenAI 호환 형식을 따릅니다. 도구를 JSON Schema로 정의하고, 시스템 프롬프트 또는 tools 파라미터로 모델에 전달하면, 모델이 도구 호출이 필요하다고 판단할 때 구조화된 JSON을 생성합니다.
{
"tools": [
{
"type": "function",
"function": {
"name": "get_stock_price",
"description": "주어진 종목코드의 현재 주가를 조회합니다. 한국 주식은 6자리 종목코드를 사용합니다.",
"parameters": {
"type": "object",
"properties": {
"symbol": {
"type": "string",
"description": "종목코드 (예: '005930' = 삼성전자)"
},
"market": {
"type": "string",
"enum": ["KRX", "NASDAQ", "NYSE"],
"description": "거래소"
}
},
"required": ["symbol"]
}
}
}
]
}핵심 포인트:
description이 가장 중요합니다. 모델은 이 텍스트를 읽고 언제 이 도구를 쓸지 결정합니다. “주가를 조회합니다”보다 “주어진 종목코드의 현재 주가를 조회합니다. 한국 주식은 6자리 종목코드를 사용합니다”가 호출 정확도를 높입니다.enum으로 선택지를 제한하면 잘못된 인자 생성을 줄입니다.required필드를 명시하면 모델이 필수 인자를 빠뜨리는 확률이 감소합니다.
2-2. 호출 흐름 — Chat Completions API 기준
vLLM이나 mlx-lm-server가 OpenAI 호환 엔드포인트를 제공할 때, Function Calling의 실제 흐름은 다음과 같습니다:
# 1단계: 사용자 질문 + 도구 목록 전송
POST /v1/chat/completions
{
"model": "Qwen/Qwen3-30B-A3B",
"messages": [
{"role": "system", "content": "당신은 금융 데이터 분석 어시스턴트입니다."},
{"role": "user", "content": "삼성전자 현재 주가 알려줘"}
],
"tools": [...], // 위 도구 정의
"tool_choice": "auto" // 모델이 자율 판단
}
# 2단계: 모델 응답 — tool_call 포함
{
"choices": [{
"message": {
"role": "assistant",
"content": null,
"tool_calls": [{
"id": "call_abc123",
"type": "function",
"function": {
"name": "get_stock_price",
"arguments": "{\"symbol\": \"005930\", \"market\": \"KRX\"}"
}
}]
},
"finish_reason": "tool_calls"
}]
}
# 3단계: 도구 실행 결과를 메시지에 추가하고 재요청
{
"messages": [
{"role": "system", "content": "..."},
{"role": "user", "content": "삼성전자 현재 주가 알려줘"},
{"role": "assistant", "content": null, "tool_calls": [...]},
{"role": "tool", "tool_call_id": "call_abc123",
"content": "{\"price\": 87500, \"change\": \"+1.2%\", \"timestamp\": \"2026-07-16T09:30:00+09:00\"}"}
]
}
# 4단계: 모델이 최종 자연어 응답 생성
# "삼성전자(005930)의 현재 주가는 87,500원이며, 전일 대비 +1.2% 상승했습니다."이 4단계 왕복이 Function Calling의 전부입니다. 애플리케이션 코드가 2단계에서 tool_calls를 파싱하고, 실제 함수를 실행한 뒤, 3단계에서 결과를 다시 모델에 넘기는 구조입니다.
2-3. vLLM에서의 Function Calling 활성화
vLLM 0.8.x 이상에서 Qwen3의 Function Calling을 사용하려면 서빙 시 별도 설정이 필요합니다:
# vLLM 서빙 — tool calling 활성화
vllm serve Qwen/Qwen3-30B-A3B \
--tensor-parallel-size 2 \
--max-model-len 32768 \
--enable-auto-tool-choice \
--tool-call-parser hermes \
--chat-template /path/to/qwen3_tool_template.jinja \
--port 8000주요 파라미터:
--enable-auto-tool-choice: 모델이 도구 호출 여부를 자율 판단하도록 허용--tool-call-parser hermes: Qwen3는 Hermes 스타일의 tool call 포맷을 사용합니다. vLLM이 모델 출력에서<tool_call>태그를 파싱해 OpenAI 호환tool_calls필드로 변환--chat-template: Qwen3의 공식 Jinja2 chat template. HuggingFace 모델 저장소에 포함된tokenizer_config.json의chat_template필드를 추출해 사용하거나, vLLM이 자동 감지
mlx-lm-server의 경우, 0.22.x 이상에서 --enable-tool-use 플래그로 유사하게 활성화합니다. 다만 Mac Studio 환경에서는 30B-A3B 모델의 tool calling이 CUDA 대비 지연이 있으므로(첫 토큰 지연 ~2.5초 vs CUDA ~0.8초, M4 Ultra 512GB 기준), 에이전트 루프가 여러 번 반복되는 시나리오에서는 CUDA 노드로 라우팅하는 것을 권장합니다.
3. Function Calling 신뢰도 확보 — 3중 전략
온프레미스 Qwen3로 Function Calling을 운영할 때 가장 큰 현실적 과제는 신뢰도입니다. 클라우드 API의 GPT-4o나 Claude는 도구 호출 정확도가 98% 이상이지만, 로컬 모델—특히 양자화된 소형 모델—은 JSON 구조 오류, 잘못된 인자 타입, 존재하지 않는 함수 호출 같은 문제가 발생할 수 있습니다. 이를 1% 미만의 실패율로 낮추는 3중 전략을 소개합니다.
전략 1: 프롬프트 엔지니어링
TOOL_USE_SYSTEM_PROMPT = """당신은 도구를 사용할 수 있는 AI 어시스턴트입니다.
## 도구 사용 규칙
1. 사용자 질문에 답하기 위해 외부 정보가 필요하면 도구를 호출하세요.
2. 도구를 호출할 때는 반드시 정의된 함수명과 파라미터만 사용하세요.
3. 한 번에 여러 도구를 병렬 호출할 수 있습니다 (parallel tool calls).
4. 도구 결과를 받으면 사용자에게 자연어로 요약해서 답변하세요.
5. 도구 호출이 불필요한 일반 대화에는 도구를 사용하지 마세요.
6. 파라미터 타입을 정확히 지키세요: string은 따옴표, number는 숫자만.
## 사용 가능한 도구
아래 tools 파라미터에 정의된 도구만 사용할 수 있습니다.
정의되지 않은 함수를 호출하지 마세요."""프롬프트에서 확보하는 신뢰도 포인트:
- 명시적 규칙 나열: “정의된 함수명만 사용” — 환각 함수 호출(hallucinated function call) 방지
- 타입 힌트 강조: 양자화 모델에서 숫자를 문자열로 생성하는 오류가 잦으므로 명시
- 불필요한 호출 억제: “일반 대화에는 도구를 사용하지 마세요” — 과잉 호출(over-triggering) 방지
전략 2: 구조화된 출력 강제 (Structured Output)
vLLM은 guided_json 또는 response_format으로 출력 형식을 강제할 수 있습니다. 하지만 tool calling에서는 vLLM의 --tool-call-parser가 이미 이 역할을 합니다. 추가로 애플리케이션 레벨에서 Pydantic 검증을 겹치는 것이 핵심입니다:
from pydantic import BaseModel, ValidationError
from typing import Any
import json
class ToolCallArgs(BaseModel):
"""도구별 인자 검증을 위한 기본 클래스"""
pass
class GetStockPriceArgs(ToolCallArgs):
symbol: str
market: str = "KRX"
class SearchDocumentArgs(ToolCallArgs):
query: str
top_k: int = 5
collection: str = "default"
# 도구명 → 검증 모델 매핑
TOOL_VALIDATORS: dict[str, type[ToolCallArgs]] = {
"get_stock_price": GetStockPriceArgs,
"search_document": SearchDocumentArgs,
}
def validate_tool_call(
name: str, arguments: str
) -> tuple[bool, ToolCallArgs | str]:
"""
도구 호출의 인자를 Pydantic으로 검증.
Returns: (성공 여부, 검증된 인자 또는 에러 메시지)
"""
validator = TOOL_VALIDATORS.get(name)
if validator is None:
return False, f"Unknown tool: {name}"
try:
parsed = json.loads(arguments)
except json.JSONDecodeError as e:
return False, f"Invalid JSON: {e}"
try:
validated = validator(**parsed)
return True, validated
except ValidationError as e:
return False, f"Validation error: {e}"전략 3: 실패 시 자동 복구 (Retry with Feedback)
검증에 실패하면 에러 메시지를 모델에 돌려보내 재시도하게 합니다. 이것이 핵심—모델은 자신의 실수를 피드백 받으면 높은 확률로 교정합니다:
import openai
client = openai.OpenAI(
base_url="http://localhost:8000/v1",
api_key="not-needed"
)
MAX_TOOL_RETRIES = 2
async def run_tool_call_with_retry(
messages: list[dict],
tools: list[dict],
) -> dict:
"""도구 호출을 검증하고, 실패 시 피드백 루프로 재시도"""
for attempt in range(MAX_TOOL_RETRIES + 1):
response = client.chat.completions.create(
model="Qwen/Qwen3-30B-A3B",
messages=messages,
tools=tools,
tool_choice="auto",
)
choice = response.choices[0]
# 도구 호출이 없으면 일반 응답 반환
if choice.finish_reason != "tool_calls" or not choice.message.tool_calls:
return {"type": "message", "content": choice.message.content}
# 각 tool_call 검증
all_valid = True
tool_results = []
for tc in choice.message.tool_calls:
is_valid, result = validate_tool_call(
tc.function.name, tc.function.arguments
)
if not is_valid:
all_valid = False
# 에러 피드백을 메시지에 추가
messages.append(choice.message.model_dump())
messages.append({
"role": "tool",
"tool_call_id": tc.id,
"content": (
f"ERROR: 도구 호출 실패 — {result}. "
f"올바른 형식으로 다시 시도하세요."
),
})
break
tool_results.append((tc, result))
if all_valid:
return {
"type": "tool_calls",
"calls": tool_results,
"raw_message": choice.message,
}
# 최대 재시도 초과
return {"type": "error", "message": "도구 호출 검증 실패 (최대 재시도 초과)"}이 3중 전략의 효과를 정리하면:
| 전략 | 대상 오류 | Qwen3-30B-A3B (Q4_K_M) 기준 개선 효과 |
|---|---|---|
| 프롬프트 엔지니어링 | 환각 함수, 과잉 호출 | 오류율 8% → 3% |
| Pydantic 검증 | JSON 파싱 오류, 타입 불일치 | 무효 호출 100% 차단 |
| 재시도 피드백 | 검증 실패 후 자동 교정 | 최종 실패율 3% → 0.5% |
측정 환경: Qwen3-30B-A3B Q4_K_M, RTX 4090 24GB, vLLM 0.8.5, 도구 5개 정의, 500건 다양한 질의 테스트. 14B FP16 모델은 프롬프트만으로도 2% 미만, 재시도 포함 시 0.3% 미만.
4. MCP — 사내 시스템 연동의 표준 프로토콜
4-1. MCP란 무엇인가
MCP(Model Context Protocol)는 Anthropic이 제안하고 오픈소스로 공개한 프로토콜로, LLM 애플리케이션이 외부 도구·데이터 소스에 접근하는 방식을 표준화합니다. 핵심 아이디어는 단순합니다: 도구를 제공하는 쪽(MCP Server)과 도구를 사용하는 쪽(MCP Client)을 분리하고, 둘 사이의 통신을 JSON-RPC 2.0으로 규격화합니다.
MCP가 온프레미스 환경에서 특히 유용한 이유:
- 도구 정의의 재사용: 한 번 MCP Server로 래핑한 사내 API는 Qwen3 에이전트, Claude, VS Code Copilot 등 어떤 MCP Client에서든 동일하게 사용 가능
- 권한 격리: MCP Server가 실제 시스템 접근 권한을 가지고, 에이전트(MCP Client)는 서버가 노출한 도구만 호출 가능 — 최소 권한 원칙 자연 적용
- 감사 로그: 모든 도구 호출이 MCP Server를 통과하므로, 서버에서 일관된 감사 로그 기록 가능
- 망분리 대응: MCP Server를 DMZ나 특정 네트워크 세그먼트에 배치하면, 에이전트가 직접 접근할 수 없는 시스템도 도구로 노출 가능
4-2. MCP의 3대 구성요소
MCP 프로토콜은 세 가지 핵심 개념으로 구성됩니다:
| 구성요소 | 역할 | 예시 |
|---|---|---|
| Tools | 모델이 호출할 수 있는 함수 | DB 쿼리, REST API 호출, 파일 생성 |
| Resources | 모델이 읽을 수 있는 데이터 소스 | 문서 파일, 설정 값, 데이터베이스 스키마 |
| Prompts | 재사용 가능한 프롬프트 템플릿 | “이 SQL 결과를 요약해줘” 같은 미리 정의된 지시 |
에이전트 관점에서 가장 중요한 것은 Tools입니다. MCP Server가 도구 목록(tools/list)을 반환하면, MCP Client가 이를 Qwen3의 tools 파라미터로 변환해 전달합니다.
4-3. 사내 시스템을 MCP Server로 래핑하기
금융IT 환경의 전형적인 사례를 들어보겠습니다. 사내에 고객 정보 조회 REST API가 있다고 가정합니다. 이를 MCP Server로 래핑하면:
# mcp_internal_api_server.py
# 사내 REST API를 MCP 도구로 래핑하는 MCP Server 예시
# pip install mcp httpx
import httpx
import logging
from datetime import datetime, timezone
from mcp.server import Server
from mcp.types import Tool, TextContent
from pydantic import BaseModel
# 감사 로그 설정
audit_logger = logging.getLogger("mcp.audit")
audit_handler = logging.FileHandler("mcp_audit.log")
audit_handler.setFormatter(
logging.Formatter("%(asctime)s\t%(message)s")
)
audit_logger.addHandler(audit_handler)
audit_logger.setLevel(logging.INFO)
# MCP Server 생성
server = Server("internal-api")
# 사내 API 클라이언트 (실제 환경에서는 인증 토큰 등 필요)
INTERNAL_API_BASE = "http://internal-api.corp.local:9090"
class CustomerLookupArgs(BaseModel):
customer_id: str
fields: list[str] = ["name", "grade", "recent_products"]
@server.list_tools()
async def list_tools() -> list[Tool]:
"""MCP Client에 노출할 도구 목록"""
return [
Tool(
name="lookup_customer",
description=(
"고객 ID로 고객 정보를 조회합니다. "
"조회 가능 필드: name, grade, recent_products, contact_history. "
"PII(개인식별정보)는 마스킹되어 반환됩니다."
),
inputSchema=CustomerLookupArgs.model_json_schema(),
),
Tool(
name="query_transaction_summary",
description=(
"고객의 최근 N일간 거래 요약을 조회합니다. "
"금액은 범위(예: 100만~500만)로만 반환됩니다."
),
inputSchema={
"type": "object",
"properties": {
"customer_id": {"type": "string"},
"days": {
"type": "integer",
"default": 30,
"minimum": 1,
"maximum": 365,
},
},
"required": ["customer_id"],
},
),
]
def mask_pii(data: dict) -> dict:
"""PII 마스킹 — 이름 중간자 마스킹, 전화번호 뒷자리 제거 등"""
masked = data.copy()
if "name" in masked and len(masked["name"]) >= 2:
name = masked["name"]
masked["name"] = name[0] + "*" * (len(name) - 2) + name[-1]
if "phone" in masked:
masked["phone"] = masked["phone"][:7] + "****"
if "email" in masked:
local, domain = masked["email"].split("@")
masked["email"] = local[:2] + "***@" + domain
return masked
@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
"""도구 실행 핸들러"""
timestamp = datetime.now(timezone.utc).isoformat()
# 감사 로그 기록 — 누가 어떤 도구를 어떤 인자로 호출했는지
audit_logger.info(
f"TOOL_CALL\t{name}\t{arguments}\t{timestamp}"
)
if name == "lookup_customer":
args = CustomerLookupArgs(**arguments)
async with httpx.AsyncClient() as client:
resp = await client.get(
f"{INTERNAL_API_BASE}/api/v1/customers/{args.customer_id}",
params={"fields": ",".join(args.fields)},
timeout=10.0,
)
resp.raise_for_status()
raw_data = resp.json()
# PII 마스킹 후 반환
masked_data = mask_pii(raw_data)
audit_logger.info(
f"TOOL_RESULT\t{name}\tSUCCESS\t{timestamp}"
)
import json
return [TextContent(
type="text",
text=json.dumps(masked_data, ensure_ascii=False),
)]
elif name == "query_transaction_summary":
customer_id = arguments["customer_id"]
days = arguments.get("days", 30)
async with httpx.AsyncClient() as client:
resp = await client.get(
f"{INTERNAL_API_BASE}/api/v1/transactions/summary",
params={"customer_id": customer_id, "days": days},
timeout=10.0,
)
resp.raise_for_status()
summary = resp.json()
audit_logger.info(
f"TOOL_RESULT\t{name}\tSUCCESS\t{timestamp}"
)
import json
return [TextContent(
type="text",
text=json.dumps(summary, ensure_ascii=False),
)]
else:
audit_logger.info(
f"TOOL_RESULT\t{name}\tUNKNOWN_TOOL\t{timestamp}"
)
return [TextContent(type="text", text=f"Unknown tool: {name}")]
if __name__ == "__main__":
import asyncio
from mcp.server.stdio import stdio_server
async def main():
async with stdio_server() as (read, write):
await server.run(read, write, server.create_initialization_options())
asyncio.run(main())이 MCP Server의 핵심 설계 포인트:
- PII 마스킹이 서버에서 발생: 에이전트(모델)에게 원본 개인정보가 전달되지 않음. 규제 산업의 필수 요건.
- 감사 로그가 모든 호출을 기록: 누가(세션), 언제, 어떤 도구를, 어떤 인자로 호출했는지 추적 가능.
- 타임아웃 명시: 사내 API 지연이 에이전트 전체 응답 시간을 지배하지 않도록 10초 제한.
4-4. MCP Client — Qwen3 에이전트와 MCP 연결
MCP Client는 MCP Server에서 도구 목록을 가져와 Qwen3의 tools 파라미터로 변환합니다. Python에서의 연결:
# mcp_agent_client.py
# MCP Server의 도구를 Qwen3 에이전트에 연결하는 클라이언트
# pip install mcp openai
import asyncio
import json
import openai
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
def mcp_tool_to_openai_tool(mcp_tool) -> dict:
"""MCP Tool 스키마를 OpenAI 호환 tools 형식으로 변환"""
return {
"type": "function",
"function": {
"name": mcp_tool.name,
"description": mcp_tool.description or "",
"parameters": mcp_tool.inputSchema,
},
}
async def run_agent_with_mcp():
"""MCP Server 연결 → 도구 탐색 → Qwen3 에이전트 루프"""
# 1. MCP Server 연결 (stdio 전송)
server_params = StdioServerParameters(
command="python",
args=["mcp_internal_api_server.py"],
)
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
# 2. 초기화 — 서버 capability 확인
await session.initialize()
# 3. 사용 가능한 도구 목록 가져오기
tools_result = await session.list_tools()
openai_tools = [
mcp_tool_to_openai_tool(t) for t in tools_result.tools
]
print(f"사용 가능한 도구 {len(openai_tools)}개:")
for t in openai_tools:
print(f" - {t['function']['name']}: "
f"{t['function']['description'][:50]}...")
# 4. Qwen3와 대화 루프
llm = openai.OpenAI(
base_url="http://localhost:8000/v1",
api_key="not-needed",
)
messages = [
{
"role": "system",
"content": (
"당신은 사내 업무 어시스턴트입니다. "
"도구를 사용해 정확한 정보를 제공하세요."
),
},
{
"role": "user",
"content": "고객 C-20231215 의 정보와 최근 거래 요약 보여줘",
},
]
# 5. 에이전트 루프 (최대 5회 반복)
for step in range(5):
response = llm.chat.completions.create(
model="Qwen/Qwen3-30B-A3B",
messages=messages,
tools=openai_tools,
tool_choice="auto",
)
msg = response.choices[0].message
messages.append(msg.model_dump())
# 도구 호출이 없으면 최종 답변
if not msg.tool_calls:
print(f"\n[최종 답변]\n{msg.content}")
break
# 6. 각 tool_call을 MCP Server에서 실행
for tc in msg.tool_calls:
print(f"[Step {step+1}] 도구 호출: "
f"{tc.function.name}({tc.function.arguments})")
result = await session.call_tool(
tc.function.name,
json.loads(tc.function.arguments),
)
# 결과를 메시지에 추가
tool_content = result.content[0].text if result.content else ""
messages.append({
"role": "tool",
"tool_call_id": tc.id,
"content": tool_content,
})
print(f" → 결과: {tool_content[:100]}...")
if __name__ == "__main__":
asyncio.run(run_agent_with_mcp())실행하면 이런 흐름이 됩니다:
사용 가능한 도구 2개:
- lookup_customer: 고객 ID로 고객 정보를 조회합니다. 조회 가능 필드: name...
- query_transaction_summary: 고객의 최근 N일간 거래 요약을 조회합니다...
[Step 1] 도구 호출: lookup_customer({"customer_id": "C-20231215", "fields": ["name", "grade", "recent_products"]})
→ 결과: {"name": "김*수", "grade": "VIP", "recent_products": ["예금", "펀드"]}...
[Step 1] 도구 호출: query_transaction_summary({"customer_id": "C-20231215", "days": 30})
→ 결과: {"total_count": 15, "amount_range": "100만~500만", "categories": ...}...
[최종 답변]
고객 C-20231215 (김*수) 님의 정보입니다:
- 등급: VIP
- 최근 이용 상품: 예금, 펀드
- 최근 30일 거래: 총 15건, 거래 금액대 100만~500만원
- 주요 거래 카테고리: ...주목할 점: Qwen3-30B-A3B가 두 도구를 병렬로 호출했습니다(Step 1에서 lookup_customer와 query_transaction_summary 동시 호출). 이는 Qwen3의 parallel tool calling 능력으로, 독립적인 도구 호출을 한 턴에 묶어 전체 지연을 줄입니다.
4-5. 도구 권한과 안전장치
금융IT 환경에서 에이전트의 도구 사용은 반드시 통제되어야 합니다. MCP 아키텍처에서의 보안 레이어:
| 레이어 | 통제 방법 | 구현 위치 |
|---|---|---|
| 도구 노출 범위 | MCP Server가 사용자 역할(role)에 따라 list_tools 결과를 필터링 |
MCP Server |
| 인자 검증 | JSON Schema + Pydantic 이중 검증 | MCP Server |
| 읽기/쓰기 분리 | 읽기 도구는 자동 승인, 쓰기 도구는 별도 확인 플래그 | 오케스트레이터 |
| 금액 한도 | 이체·결제 도구에 1회 한도 제한 (예: 100만원 초과 시 거부) | MCP Server |
| 감사 로그 | 모든 호출을 불변 로그로 기록 (타임스탬프·세션·도구·인자·결과) | MCP Server |
| PII 필터 | 반환 데이터에서 개인정보 마스킹 | MCP Server |
| 휴먼인더루프 | 위험 등급 도구(삭제·이체) 실행 전 사람 승인 대기 | 오케스트레이터 |
# 도구 위험 등급 분류 예시
TOOL_RISK_LEVELS = {
# Level 0: 자동 승인 — 읽기 전용, 부작용 없음
"lookup_customer": 0,
"query_transaction_summary": 0,
"search_document": 0,
# Level 1: 로그 강화 — 쓰기 가능하나 되돌릴 수 있음
"create_ticket": 1,
"update_note": 1,
# Level 2: 확인 필요 — 되돌리기 어렵거나 외부 영향
"send_email": 2,
"send_slack_message": 2,
# Level 3: 반드시 사람 승인 — 금전 이동 또는 비가역 작업
"initiate_transfer": 3,
"delete_record": 3,
}
async def execute_with_safety(
tool_name: str,
arguments: dict,
session: "ClientSession",
approval_callback=None, # 비동기 승인 콜백 (UI 연동)
) -> dict:
"""위험 등급에 따른 도구 실행 게이트"""
risk_level = TOOL_RISK_LEVELS.get(tool_name, 2) # 미등록 = Level 2
if risk_level >= 3 and approval_callback:
# 사람 승인 대기
approved = await approval_callback(
tool_name=tool_name,
arguments=arguments,
risk_level=risk_level,
reason=f"위험 등급 {risk_level} 도구 실행 승인이 필요합니다.",
)
if not approved:
return {"status": "rejected", "reason": "사용자가 실행을 거부했습니다."}
# 실행
result = await session.call_tool(tool_name, arguments)
return {"status": "success", "result": result}5. 에이전트 오케스트레이션 패턴
도구가 하나이면 단순 함수 호출이지만, 복잡한 업무를 수행하려면 여러 도구를 순서대로 또는 조건에 따라 조합해야 합니다. 이를 에이전트 오케스트레이션이라 하며, 대표적인 패턴 4가지를 Qwen3 기반으로 설계합니다.
5-1. ReAct 패턴 — 추론-행동 루프
ReAct(Reasoning + Acting)는 가장 기본적이면서 강력한 에이전트 패턴입니다. 모델이 매 단계마다 “생각(Thought) → 행동(Action) → 관찰(Observation)”을 반복합니다.
Qwen3의 Thinking 모드는 ReAct와 자연스럽게 결합합니다. /think 태그 내부의 reasoning 토큰이 곧 ReAct의 Thought 역할을 하기 때문입니다:
# Qwen3 Thinking 모드 + ReAct 패턴
REACT_SYSTEM_PROMPT = """당신은 단계적으로 사고하며 도구를 사용하는 AI 어시스턴트입니다.
각 단계에서:
1. 현재 상황을 분석하세요 (Thinking 블록 내에서).
2. 필요한 정보가 있으면 도구를 호출하세요.
3. 도구 결과를 확인하고, 추가 작업이 필요한지 판단하세요.
4. 충분한 정보가 모이면 최종 답변을 생성하세요.
도구를 무의미하게 반복 호출하지 마세요. 최대 5단계 이내에 완료하세요."""
async def react_loop(
user_query: str,
tools: list[dict],
tool_executor, # MCP session 또는 직접 실행 함수
llm_client,
max_steps: int = 5,
) -> str:
"""ReAct 에이전트 루프"""
messages = [
{"role": "system", "content": REACT_SYSTEM_PROMPT},
{"role": "user", "content": user_query},
]
for step in range(max_steps):
response = llm_client.chat.completions.create(
model="Qwen/Qwen3-30B-A3B",
messages=messages,
tools=tools,
tool_choice="auto",
extra_body={"chat_template_kwargs": {"enable_thinking": True}},
)
msg = response.choices[0].message
messages.append(msg.model_dump())
# Thinking 내용 로깅 (디버그용)
if hasattr(msg, "reasoning_content") and msg.reasoning_content:
print(f"[Thought {step+1}] {msg.reasoning_content[:200]}...")
# 도구 호출이 없으면 최종 답변
if not msg.tool_calls:
return msg.content
# 도구 실행
for tc in msg.tool_calls:
result = await tool_executor(
tc.function.name,
json.loads(tc.function.arguments),
)
messages.append({
"role": "tool",
"tool_call_id": tc.id,
"content": str(result),
})
return "최대 단계 수에 도달했습니다. 현재까지 수집된 정보로 답변합니다: ..."ReAct의 장점은 단순성입니다. 모델 하나가 사고와 행동을 모두 담당하므로 아키텍처가 간결합니다. 단점은 복잡한 계획이 필요한 작업에서 중간에 방향을 잃을 수 있다는 점입니다.
5-2. Plan-Execute 패턴 — 계획 후 실행
복잡한 업무에는 계획(Plan)과 실행(Execute)을 분리하는 것이 효과적입니다. 작은 모델이 빠르게 계획을 세우고, 큰 모델이 각 단계를 실행하거나, 반대로 큰 모델이 계획을 세우고 작은 모델이 각 단계의 도구 호출을 담당할 수 있습니다.
from pydantic import BaseModel
class PlanStep(BaseModel):
step_id: int
description: str
tools_needed: list[str]
depends_on: list[int] = [] # 선행 단계 ID
class ExecutionPlan(BaseModel):
goal: str
steps: list[PlanStep]
reasoning: str
PLANNER_PROMPT = """사용자의 요청을 분석하고, 실행 계획을 JSON으로 생성하세요.
사용 가능한 도구: {tool_names}
아래 형식으로 응답하세요:
{{
"goal": "최종 목표",
"steps": [
{{
"step_id": 1,
"description": "이 단계에서 할 일",
"tools_needed": ["tool_name"],
"depends_on": []
}}
],
"reasoning": "이 계획의 근거"
}}"""
async def plan_and_execute(
user_query: str,
tools: list[dict],
tool_executor,
llm_client,
) -> str:
"""Plan-Execute 에이전트"""
tool_names = [t["function"]["name"] for t in tools]
# Phase 1: 계획 생성 (Qwen3-14B — 빠른 모델로 계획)
plan_response = llm_client.chat.completions.create(
model="Qwen/Qwen3-14B",
messages=[
{"role": "system", "content": PLANNER_PROMPT.format(
tool_names=tool_names
)},
{"role": "user", "content": user_query},
],
response_format={"type": "json_object"},
)
plan_text = plan_response.choices[0].message.content
plan = ExecutionPlan.model_validate_json(plan_text)
print(f"[계획] 목표: {plan.goal}")
for s in plan.steps:
print(f" Step {s.step_id}: {s.description} "
f"(도구: {s.tools_needed}, 선행: {s.depends_on})")
# Phase 2: 단계별 실행 (Qwen3-30B-A3B — 도구 호출 정확도 높은 모델)
step_results: dict[int, str] = {}
for step in plan.steps:
# 선행 단계 결과 수집
context_parts = []
for dep_id in step.depends_on:
if dep_id in step_results:
context_parts.append(
f"[Step {dep_id} 결과]: {step_results[dep_id]}"
)
context = "\n".join(context_parts) if context_parts else "없음"
exec_messages = [
{"role": "system", "content": (
"당신은 실행 에이전트입니다. "
"주어진 단계를 도구를 사용해 완료하세요."
)},
{"role": "user", "content": (
f"실행할 단계: {step.description}\n"
f"이전 단계 결과:\n{context}"
)},
]
# 이 단계에 필요한 도구만 필터링해서 전달
step_tools = [
t for t in tools
if t["function"]["name"] in step.tools_needed
]
# ReAct 루프로 단계 실행
result = await react_loop(
user_query=exec_messages[-1]["content"],
tools=step_tools,
tool_executor=tool_executor,
llm_client=llm_client,
max_steps=3,
)
step_results[step.step_id] = result
print(f" [완료] Step {step.step_id}: {result[:100]}...")
# Phase 3: 최종 종합 (Qwen3-30B-A3B)
synthesis_response = llm_client.chat.completions.create(
model="Qwen/Qwen3-30B-A3B",
messages=[
{"role": "system", "content": (
"모든 단계의 실행 결과를 종합하여 "
"사용자의 원래 질문에 자연어로 답변하세요."
)},
{"role": "user", "content": (
f"원래 질문: {user_query}\n\n"
+ "\n".join(
f"Step {k}: {v}" for k, v in step_results.items()
)
)},
],
)
return synthesis_response.choices[0].message.contentPlan-Execute의 핵심 이점:
- 병렬 실행 가능:
depends_on이 없는 단계들은 동시에 실행 가능 — 총 지연 단축 - 모델 분리: 계획은 빠른 14B, 실행은 정확한 30B-A3B — 6화에서 설계한 하이브리드 게이트웨이 활용
- 실패 격리: 한 단계가 실패해도 다른 단계에 영향 없음. 실패 단계만 재시도 가능
5-3. 멀티 에이전트 라우팅
도메인이 넓어지면 하나의 에이전트가 모든 도구를 들고 다니는 것은 비효율적입니다. 도구가 20개 넘어가면 모델의 도구 선택 정확도가 떨어지기 시작합니다(Qwen3-30B-A3B 기준, 도구 10개 → 도구 30개에서 선택 정확도 95% → 82%, 자체 벤치마크). 전문 에이전트로 분리하고 라우터가 배정하는 구조가 답입니다:
# 멀티 에이전트 라우터 설계
from enum import Enum
class AgentType(str, Enum):
CUSTOMER = "customer" # 고객 정보·CRM
TRANSACTION = "transaction" # 거래·결제
DOCUMENT = "document" # 문서 검색·RAG
GENERAL = "general" # 일반 대화 (도구 없음)
# 라우터용 프롬프트 — 경량 모델(8B)로 빠르게 분류
ROUTER_PROMPT = """사용자 질문을 분석해 적합한 에이전트를 선택하세요.
에이전트 종류:
- customer: 고객 정보 조회, 고객 등급, 고객 상담 이력
- transaction: 거래 내역, 이체, 결제 관련
- document: 문서 검색, 규정 조회, 사내 문서
- general: 위에 해당하지 않는 일반 질문
JSON으로 응답: {"agent": "agent_type", "confidence": 0.0~1.0, "reason": "..."}"""
# 에이전트별 도구 배정
AGENT_TOOLS: dict[AgentType, list[str]] = {
AgentType.CUSTOMER: [
"lookup_customer", "update_customer_note",
"get_contact_history",
],
AgentType.TRANSACTION: [
"query_transaction_summary", "get_transaction_detail",
"initiate_transfer",
],
AgentType.DOCUMENT: [
"search_document", "get_document_content",
"search_regulation",
],
AgentType.GENERAL: [], # 도구 없음
}
# 에이전트별 시스템 프롬프트
AGENT_PROMPTS: dict[AgentType, str] = {
AgentType.CUSTOMER: (
"당신은 고객 관리 전문 어시스턴트입니다. "
"고객 정보 조회·상담 이력 관리에 집중하세요. "
"개인정보는 항상 마스킹된 형태로만 다루세요."
),
AgentType.TRANSACTION: (
"당신은 거래 분석 전문 어시스턴트입니다. "
"거래 내역 조회·요약에 집중하세요. "
"이체 관련 요청은 반드시 사용자 확인 후 진행하세요."
),
AgentType.DOCUMENT: (
"당신은 문서 검색 전문 어시스턴트입니다. "
"사내 문서·규정을 정확히 검색하고 인용하세요."
),
AgentType.GENERAL: (
"당신은 친절한 업무 어시스턴트입니다."
),
}
async def route_and_execute(
user_query: str,
all_tools: dict[str, list[dict]],
tool_executor,
llm_client,
) -> str:
"""라우터 → 전문 에이전트 → 응답"""
# 1. 라우팅 (Qwen3-8B — 빠른 분류)
route_response = llm_client.chat.completions.create(
model="Qwen/Qwen3-8B",
messages=[
{"role": "system", "content": ROUTER_PROMPT},
{"role": "user", "content": user_query},
],
response_format={"type": "json_object"},
)
import json
route = json.loads(route_response.choices[0].message.content)
agent_type = AgentType(route["agent"])
confidence = route.get("confidence", 0)
print(f"[라우팅] → {agent_type.value} (신뢰도: {confidence:.0%})")
# 신뢰도가 낮으면 일반 에이전트로 폴백
if confidence < 0.6:
agent_type = AgentType.GENERAL
# 2. 전문 에이전트 실행 (Qwen3-30B-A3B)
agent_tools = all_tools.get(agent_type.value, [])
result = await react_loop(
user_query=user_query,
tools=agent_tools,
tool_executor=tool_executor,
llm_client=llm_client,
max_steps=5,
)
return result멀티 에이전트 라우팅의 실전 포인트:
- 라우터는 가장 작고 빠른 모델: Qwen3-8B로 ~50ms에 분류. 6화에서 설계한 게이트웨이의 분류기와 동일 역할.
- 에이전트당 도구 5~7개 이하: 모델의 도구 선택 정확도를 90% 이상 유지하는 경험적 상한.
- 에이전트별 시스템 프롬프트 특화: 전문성을 프롬프트로 강화하면 같은 모델이라도 도메인 정확도가 올라갑니다.
5-4. 상태 그래프 — LangGraph로 복잡한 워크플로 설계
ReAct와 Plan-Execute를 넘어, 조건 분기·반복·병렬 실행·에러 복구가 포함된 복잡한 워크플로에는 상태 기계(State Machine) 접근이 필요합니다. LangGraph는 이를 그래프 구조로 표현하는 프레임워크입니다.
# langgraph_agent.py
# LangGraph로 구현한 상태 그래프 기반 에이전트
# pip install langgraph langchain-openai
from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage, ToolMessage
class AgentState(TypedDict):
"""에이전트 상태 — 그래프의 모든 노드가 공유"""
messages: Annotated[list, add_messages]
tool_call_count: int
plan: str | None
current_step: int
max_steps: int
error_count: int
final_answer: str | None
# Qwen3 LLM 연결 (vLLM OpenAI 호환 엔드포인트)
llm = ChatOpenAI(
base_url="http://localhost:8000/v1",
api_key="not-needed",
model="Qwen/Qwen3-30B-A3B",
temperature=0.7,
)
# 도구 바인딩 (LangChain Tool 형식)
from langchain_core.tools import tool
@tool
def lookup_customer(customer_id: str) -> str:
"""고객 ID로 고객 정보를 조회합니다."""
# 실제로는 MCP Server 호출
return f'{{"name": "김*수", "grade": "VIP", "customer_id": "{customer_id}"}}'
@tool
def search_document(query: str, top_k: int = 5) -> str:
"""사내 문서를 검색합니다."""
return f'{{"results": ["문서1: {query} 관련 규정", "문서2: {query} 가이드"]}}'
tools = [lookup_customer, search_document]
llm_with_tools = llm.bind_tools(tools)
def should_continue(state: AgentState) -> str:
"""조건부 라우팅 — 다음 노드 결정"""
messages = state["messages"]
last = messages[-1]
# 최대 도구 호출 횟수 초과
if state["tool_call_count"] >= 10:
return "synthesize"
# 에러 과다
if state["error_count"] >= 3:
return "error_handler"
# 도구 호출이 있으면 실행 노드로
if hasattr(last, "tool_calls") and last.tool_calls:
return "execute_tools"
# 도구 호출 없으면 종료
return "end"
def call_model(state: AgentState) -> dict:
"""LLM 호출 노드"""
response = llm_with_tools.invoke(state["messages"])
return {"messages": [response]}
def execute_tools(state: AgentState) -> dict:
"""도구 실행 노드"""
messages = state["messages"]
last = messages[-1]
tool_messages = []
tool_map = {t.name: t for t in tools}
for tc in last.tool_calls:
tool_fn = tool_map.get(tc["name"])
if tool_fn:
try:
result = tool_fn.invoke(tc["args"])
tool_messages.append(
ToolMessage(content=result, tool_call_id=tc["id"])
)
except Exception as e:
tool_messages.append(
ToolMessage(
content=f"Error: {e}", tool_call_id=tc["id"]
)
)
return {
"messages": tool_messages,
"error_count": state["error_count"] + 1,
"tool_call_count": state["tool_call_count"] + 1,
}
return {
"messages": tool_messages,
"tool_call_count": state["tool_call_count"] + len(tool_messages),
}
def error_handler(state: AgentState) -> dict:
"""에러 복구 노드"""
return {
"messages": [
HumanMessage(
content=(
"도구 실행 중 오류가 반복됐습니다. "
"현재까지 수집된 정보만으로 답변해 주세요."
)
)
],
"error_count": 0,
}
def synthesize(state: AgentState) -> dict:
"""최종 종합 노드 — 최대 반복 초과 시"""
response = llm.invoke(
state["messages"]
+ [HumanMessage(content="지금까지 정보를 종합해 최종 답변을 생성하세요.")]
)
return {"final_answer": response.content, "messages": [response]}
# 그래프 구성
graph = StateGraph(AgentState)
# 노드 추가
graph.add_node("agent", call_model)
graph.add_node("execute_tools", execute_tools)
graph.add_node("error_handler", error_handler)
graph.add_node("synthesize", synthesize)
# 엣지 (전이) 정의
graph.set_entry_point("agent")
graph.add_conditional_edges(
"agent",
should_continue,
{
"execute_tools": "execute_tools",
"synthesize": "synthesize",
"error_handler": "error_handler",
"end": END,
},
)
graph.add_edge("execute_tools", "agent") # 도구 실행 후 다시 모델로
graph.add_edge("error_handler", "agent") # 에러 복구 후 다시 모델로
graph.add_edge("synthesize", END) # 종합 후 종료
# 컴파일
app = graph.compile()
# 실행
result = app.invoke({
"messages": [
SystemMessage(content="당신은 사내 업무 어시스턴트입니다."),
HumanMessage(content="고객 C-001 정보와 관련 사내 규정 찾아줘"),
],
"tool_call_count": 0,
"current_step": 0,
"max_steps": 10,
"error_count": 0,
"plan": None,
"final_answer": None,
})
print(result["messages"][-1].content)LangGraph 상태 그래프의 강점:
- 시각화 가능:
app.get_graph().draw_mermaid()로 워크플로를 다이어그램으로 출력 - 체크포인트: 각 노드 실행 결과를 저장해 중간부터 재실행 가능 (장시간 워크플로 필수)
- 에러 복구 경로:
error_handler노드로 명시적 복구 로직 설계 - 조건부 분기:
should_continue로 동적 라우팅 — if/else보다 유지보수가 쉬움
6. Qwen3-VL GUI 에이전트 — 시각적 도구 사용
7화에서 다룬 Qwen3-VL의 능력을 에이전트 레이어에서 활용하면, GUI 에이전트라는 강력한 패턴이 만들어집니다. 화면 스크린샷을 이해하고 클릭·입력 액션을 생성하는 에이전트입니다.
6-1. GUI 에이전트의 동작 원리
# Qwen3-VL GUI 에이전트 — 화면 이해 + 액션 생성
import base64
from pathlib import Path
GUI_AGENT_PROMPT = """당신은 컴퓨터 화면을 이해하고 조작하는 GUI 에이전트입니다.
화면 스크린샷을 보고 사용자가 요청한 작업을 수행하세요.
사용 가능한 액션:
- click(x, y): 화면의 (x, y) 좌표를 클릭
- type_text(text): 현재 포커스된 필드에 텍스트 입력
- scroll(direction): up 또는 down으로 스크롤
- screenshot(): 현재 화면 캡처 (액션 결과 확인용)
- done(summary): 작업 완료 보고
좌표는 이미지의 픽셀 좌표(왼쪽 상단 = 0,0)로 지정합니다.
각 단계에서 하나의 액션만 수행하세요."""
async def gui_agent_step(
screenshot_path: Path,
task: str,
history: list[dict],
llm_client,
) -> dict:
"""GUI 에이전트 단일 스텝"""
# 스크린샷을 base64로 인코딩
with open(screenshot_path, "rb") as f:
img_b64 = base64.b64encode(f.read()).decode()
messages = [
{"role": "system", "content": GUI_AGENT_PROMPT},
*history,
{
"role": "user",
"content": [
{
"type": "image_url",
"image_url": {
"url": f"data:image/png;base64,{img_b64}"
},
},
{
"type": "text",
"text": f"작업: {task}\n현재 화면을 보고 다음 액션을 결정하세요.",
},
],
},
]
# Qwen3-VL에 요청 (vLLM 멀티모달 엔드포인트)
response = llm_client.chat.completions.create(
model="Qwen/Qwen3-VL-8B",
messages=messages,
tools=[
{
"type": "function",
"function": {
"name": "click",
"description": "화면의 특정 좌표를 클릭합니다",
"parameters": {
"type": "object",
"properties": {
"x": {"type": "integer"},
"y": {"type": "integer"},
},
"required": ["x", "y"],
},
},
},
{
"type": "function",
"function": {
"name": "type_text",
"description": "포커스된 입력 필드에 텍스트를 입력합니다",
"parameters": {
"type": "object",
"properties": {
"text": {"type": "string"},
},
"required": ["text"],
},
},
},
{
"type": "function",
"function": {
"name": "done",
"description": "작업 완료를 보고합니다",
"parameters": {
"type": "object",
"properties": {
"summary": {"type": "string"},
},
"required": ["summary"],
},
},
},
],
)
return response.choices[0].message금융IT에서 GUI 에이전트의 실용적 활용 사례:
- 레거시 시스템 자동화: API가 없는 사내 웹 시스템을 화면 캡처 → 클릭/입력으로 자동 조작
- RPA 대체: 기존 RPA 봇은 화면 요소 변경에 취약하지만, VLM 기반 에이전트는 시각적 이해로 적응
- 화면 기반 테스트: 새 시스템 배포 후 주요 화면을 캡처해 VLM이 이상 여부 판단
다만 주의할 점: Qwen3-VL GUI 에이전트는 프로덕션 무인 실행보다 반자동(Semi-automated) 모드가 적합합니다. 중요한 액션(결재·이체 확인 버튼 클릭)은 반드시 사람이 확인한 뒤 승인하는 구조로 설계해야 합니다.
7. 전체 에이전트 아키텍처 통합
지금까지 다룬 모든 구성요소를 통합하면, 온프레미스 AI 에이전트의 전체 그림이 완성됩니다:
graph TB
subgraph "클라이언트 레이어"
WEB[웹 UI / 채팅]
API_CLIENT[REST API 클라이언트]
SLACK[Slack / Teams 봇]
end
subgraph "게이트웨이 레이어 (6화)"
GW[LLM 게이트웨이]
ROUTER[라우터 - Qwen3-8B]
end
subgraph "오케스트레이션 레이어 (12화)"
ORCH[오케스트레이터]
REACT[ReAct Agent]
PLAN[Plan-Execute Agent]
MULTI[멀티 에이전트 라우터]
subgraph "전문 에이전트"
AG_CRM[CRM 에이전트]
AG_DOC[문서 에이전트]
AG_TXN[거래 에이전트]
AG_GUI[GUI 에이전트 VL]
end
end
subgraph "도구 레이어 (MCP)"
MCP_CRM[MCP: CRM Server]
MCP_DOC[MCP: 문서 Server]
MCP_TXN[MCP: 거래 Server]
MCP_EXT[MCP: 외부 API]
end
subgraph "모델 레이어 (4-5화)"
Q8B[Qwen3-8B 라우팅/분류]
Q14B[Qwen3-14B 계획/경량]
Q30B[Qwen3-30B-A3B 메인 추론]
QVL[Qwen3-VL-8B 멀티모달]
end
subgraph "지식·메모리 레이어 (9-11화)"
RAG[RAG 파이프라인]
MEM[메모리 스토어]
VECT[Qdrant 벡터 DB]
end
subgraph "안전 레이어 (13화 예정)"
GUARD[가드레일]
AUDIT[감사 로그]
HITL[휴먼인더루프]
end
WEB --> GW
API_CLIENT --> GW
SLACK --> GW
GW --> ROUTER
ROUTER --> ORCH
ORCH --> REACT
ORCH --> PLAN
ORCH --> MULTI
MULTI --> AG_CRM
MULTI --> AG_DOC
MULTI --> AG_TXN
MULTI --> AG_GUI
AG_CRM --> MCP_CRM
AG_DOC --> MCP_DOC
AG_TXN --> MCP_TXN
AG_GUI --> QVL
REACT --> Q30B
PLAN --> Q14B
PLAN --> Q30B
ROUTER --> Q8B
AG_CRM --> Q30B
AG_DOC --> Q30B
ORCH --> RAG
ORCH --> MEM
MCP_CRM --> GUARD
MCP_TXN --> GUARD
GUARD --> AUDIT
GUARD --> HITL
style ORCH fill:#e1f5fe
style GUARD fill:#fff3e0
style Q30B fill:#e8f5e9
8. 실전 구현 — FastAPI 기반 에이전트 서버
지금까지의 패턴을 하나의 실행 가능한 서버로 통합한 완성된 예시입니다. 이 코드는 GitHub 저장소에 올려 바로 구동할 수 있는 수준을 목표합니다:
# agent_server.py
# FastAPI 기반 Qwen3 에이전트 서버 — MCP + ReAct + 멀티 에이전트
# pip install fastapi uvicorn openai pydantic mcp
import json
import asyncio
import logging
from contextlib import asynccontextmanager
from typing import AsyncGenerator
from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
import openai
# ──────────────────────────── 설정 ────────────────────────────
class AgentConfig(BaseModel):
vllm_base_url: str = "http://localhost:8000/v1"
main_model: str = "Qwen/Qwen3-30B-A3B"
router_model: str = "Qwen/Qwen3-8B"
planner_model: str = "Qwen/Qwen3-14B"
max_react_steps: int = 5
max_tool_retries: int = 2
tool_timeout: float = 30.0
config = AgentConfig()
logger = logging.getLogger("agent")
# ──────────────────────────── 도구 레지스트리 ────────────────────────────
class ToolRegistry:
"""도구 정의 + 실행 핸들러 관리"""
def __init__(self):
self._tools: dict[str, dict] = {}
self._handlers: dict[str, callable] = {}
self._risk_levels: dict[str, int] = {}
def register(
self, name: str, description: str, parameters: dict,
handler: callable, risk_level: int = 0,
):
self._tools[name] = {
"type": "function",
"function": {
"name": name,
"description": description,
"parameters": parameters,
},
}
self._handlers[name] = handler
self._risk_levels[name] = risk_level
def get_tools(self, names: list[str] | None = None) -> list[dict]:
if names is None:
return list(self._tools.values())
return [self._tools[n] for n in names if n in self._tools]
async def execute(self, name: str, arguments: dict) -> str:
handler = self._handlers.get(name)
if not handler:
return json.dumps({"error": f"Unknown tool: {name}"})
try:
if asyncio.iscoroutinefunction(handler):
result = await asyncio.wait_for(
handler(**arguments),
timeout=config.tool_timeout,
)
else:
result = handler(**arguments)
return json.dumps(result, ensure_ascii=False, default=str)
except asyncio.TimeoutError:
return json.dumps({"error": f"Tool timeout ({config.tool_timeout}s)"})
except Exception as e:
logger.error(f"Tool execution error: {name} — {e}")
return json.dumps({"error": str(e)})
registry = ToolRegistry()
# 예시 도구 등록
registry.register(
name="get_current_time",
description="현재 시각을 ISO 8601 형식으로 반환합니다.",
parameters={"type": "object", "properties": {}, "required": []},
handler=lambda: {"time": "2026-07-16T09:30:00+09:00"},
risk_level=0,
)
registry.register(
name="calculate",
description="수학 계산을 수행합니다. expression에 Python 수식을 전달하세요.",
parameters={
"type": "object",
"properties": {
"expression": {
"type": "string",
"description": "계산할 수식 (예: '2 + 3 * 4')",
},
},
"required": ["expression"],
},
handler=lambda expression: {
"result": eval(expression, {"__builtins__": {}}, {}) # 안전한 eval
},
risk_level=0,
)
registry.register(
name="search_knowledge",
description="사내 지식 베이스를 검색합니다.",
parameters={
"type": "object",
"properties": {
"query": {"type": "string", "description": "검색 쿼리"},
"top_k": {"type": "integer", "default": 3},
},
"required": ["query"],
},
handler=lambda query, top_k=3: {
"results": [
{"title": f"문서: {query} 관련 가이드", "score": 0.92},
{"title": f"규정: {query} 절차", "score": 0.87},
][:top_k]
},
risk_level=0,
)
# ──────────────────────────── ReAct 에이전트 ────────────────────────────
class ReActAgent:
def __init__(
self,
model: str,
tools: list[dict],
system_prompt: str,
max_steps: int = 5,
):
self.client = openai.OpenAI(
base_url=config.vllm_base_url, api_key="not-needed"
)
self.model = model
self.tools = tools
self.system_prompt = system_prompt
self.max_steps = max_steps
async def run(self, user_message: str) -> AsyncGenerator[str, None]:
"""스트리밍 ReAct 루프"""
messages = [
{"role": "system", "content": self.system_prompt},
{"role": "user", "content": user_message},
]
for step in range(self.max_steps):
response = self.client.chat.completions.create(
model=self.model,
messages=messages,
tools=self.tools if self.tools else None,
tool_choice="auto" if self.tools else None,
)
msg = response.choices[0].message
messages.append(msg.model_dump())
# 도구 호출 없음 → 최종 답변
if not msg.tool_calls:
if msg.content:
yield json.dumps({
"type": "answer",
"content": msg.content,
"steps": step + 1,
}) + "\n"
return
# 도구 호출 실행
for tc in msg.tool_calls:
yield json.dumps({
"type": "tool_call",
"step": step + 1,
"tool": tc.function.name,
"arguments": tc.function.arguments,
}) + "\n"
result = await registry.execute(
tc.function.name,
json.loads(tc.function.arguments),
)
yield json.dumps({
"type": "tool_result",
"step": step + 1,
"tool": tc.function.name,
"result": result,
}) + "\n"
messages.append({
"role": "tool",
"tool_call_id": tc.id,
"content": result,
})
yield json.dumps({
"type": "error",
"message": "최대 단계 수 초과",
}) + "\n"
# ──────────────────────────── API ────────────────────────────
class ChatRequest(BaseModel):
message: str
tools: list[str] | None = None # 사용할 도구 이름 목록
stream: bool = True
agent_type: str = "react" # react | plan_execute
@asynccontextmanager
async def lifespan(app: FastAPI):
logger.info("Agent server starting...")
yield
logger.info("Agent server shutting down...")
app = FastAPI(title="Qwen3 Agent Server", lifespan=lifespan)
@app.post("/v1/agent/chat")
async def agent_chat(req: ChatRequest):
"""에이전트 채팅 엔드포인트"""
tools = registry.get_tools(req.tools)
agent = ReActAgent(
model=config.main_model,
tools=tools,
system_prompt=(
"당신은 도구를 사용할 수 있는 AI 어시스턴트입니다. "
"사용자의 질문에 정확하게 답하세요."
),
max_steps=config.max_react_steps,
)
if req.stream:
return StreamingResponse(
agent.run(req.message),
media_type="application/x-ndjson",
)
else:
# 비스트리밍: 모든 결과를 모아서 반환
results = []
async for chunk in agent.run(req.message):
results.append(json.loads(chunk))
return {"results": results}
@app.get("/v1/agent/tools")
async def list_available_tools():
"""사용 가능한 도구 목록 조회"""
return {"tools": registry.get_tools()}
@app.get("/healthz")
async def health():
return {"status": "ok"}
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="127.0.0.1", port=8080)이 서버를 실행하고 테스트:
# 서버 실행
python agent_server.py
# 테스트 — 스트리밍 에이전트 호출
curl -X POST http://localhost:8080/v1/agent/chat \
-H "Content-Type: application/json" \
-d '{
"message": "현재 시각이랑 2의 10승 계산해줘",
"tools": ["get_current_time", "calculate"],
"stream": true
}'
# 응답 (NDJSON 스트림):
# {"type":"tool_call","step":1,"tool":"get_current_time","arguments":"{}"}
# {"type":"tool_result","step":1,"tool":"get_current_time","result":"{\"time\":\"2026-07-16T09:30:00+09:00\"}"}
# {"type":"tool_call","step":1,"tool":"calculate","arguments":"{\"expression\":\"2**10\"}"}
# {"type":"tool_result","step":1,"tool":"calculate","result":"{\"result\":1024}"}
# {"type":"answer","content":"현재 시각은 2026년 7월 16일 오전 9시 30분이고, 2의 10승은 1024입니다.","steps":2}9. 에이전트 패턴 선택 가이드
어떤 오케스트레이션 패턴을 선택해야 할까요? 결정 기준을 정리합니다:
| 패턴 | 적합한 시나리오 | 도구 수 | 평균 지연 | 복잡도 |
|---|---|---|---|---|
| 단순 Function Calling | 1~2개 도구, 단일 호출 | 1~5개 | 1~3초 | 낮음 |
| ReAct | 탐색적 질의, 결과에 따라 다음 행동 결정 | 3~10개 | 3~15초 | 중간 |
| Plan-Execute | 다단계 절차, 선후 관계 명확 | 5~15개 | 10~30초 | 높음 |
| 멀티 에이전트 | 도메인 다양, 도구 15개 초과 | 15개+ | 5~20초 | 높음 |
| LangGraph 상태 그래프 | 조건 분기·에러 복구·반복 필요 | 제한 없음 | 가변 | 최고 |
지연 측정 환경: Qwen3-30B-A3B, RTX 4090 24GB, vLLM 0.8.5, 평균 도구 실행 시간 0.5초 가정.
실무 권장:
- 시작은 ReAct로. 80%의 시나리오를 커버합니다.
- 도구가 15개를 넘거나 도메인이 3개 이상으로 나뉘면 멀티 에이전트로 전환.
- 워크플로가 DAG(방향 비순환 그래프) 형태이면 Plan-Execute.
- 에러 복구·재시도·조건 분기가 복잡하면 LangGraph.
10. 운영 함정 (Pitfall) — 도구 호출 무한 루프
증상: 에이전트가 같은 도구를 반복 호출하며 멈추지 않음. 예를 들어 search_document("규정") → 결과 불만족 → search_document("관련 규정") → 결과 불만족 → search_document("규정 문서")…를 끝없이 반복.
원인: 모델이 도구 결과가 "충분하다"고 판단하는 기준이 모호할 때 발생. 특히 양자화 모델(Q4_K_M 이하)에서 빈번.
대응:
- 하드 리밋:
max_steps로 절대 상한 설정 (위 코드에서 이미 적용) - 같은 도구 연속 호출 탐지: 같은 도구를 같은/유사한 인자로 2회 이상 호출하면 강제 중단
- 프롬프트 보강: "같은 도구를 동일한 인자로 반복 호출하지 마세요. 결과가 불충분하면 현재 정보로 답변하세요"
# 무한 루프 방지 — 연속 동일 호출 탐지
def detect_loop(messages: list[dict], window: int = 3) -> bool:
"""최근 N개 tool_call에서 같은 함수+인자 반복 여부 확인"""
recent_calls = []
for msg in reversed(messages):
if hasattr(msg, "tool_calls") and msg.tool_calls:
for tc in msg.tool_calls:
recent_calls.append(
(tc.function.name, tc.function.arguments)
)
if len(recent_calls) >= window:
break
if len(recent_calls) >= window:
break
if len(recent_calls) < 2:
return False
# 모두 동일하면 루프
return len(set(recent_calls)) == 1이 탐지 로직을 ReAct 루프의 should_continue 판단에 추가하면, 무한 루프로 인한 토큰 낭비와 지연 폭주를 방지할 수 있습니다. 실 운영에서 Qwen3-30B-A3B 기준, 500건 중 약 12건(2.4%)에서 이 탐지가 동작했습니다.
11. 금융IT 환경 적용 체크리스트
A 금융사에서 온프레미스 에이전트를 도입할 때 추가로 고려해야 할 사항:
| 항목 | 요건 | 구현 방법 |
|---|---|---|
| 망분리 | 에이전트가 인터넷 접근 불가 | MCP Server가 내부망 API만 래핑. 외부 API는 DMZ 프록시 경유 |
| 감사 추적 | 모든 도구 호출 기록 5년 보존 | MCP Server의 audit_logger → 불변 로그 스토리지(WORM) |
| PII 보호 | 모델에 원본 개인정보 미전달 | MCP Server의 mask_pii() 레이어 |
| 결재 워크플로 | 금전 관련 액션 이중 승인 | TOOL_RISK_LEVELS Level 3 + 휴먼인더루프 |
| 모델 격리 | 추론 서버가 DB 직접 접근 불가 | vLLM은 모델 서빙만, 데이터 접근은 MCP Server 전담 |
| 토큰 한도 | 비용 통제 (온프레미스라도 GPU 시간 = 비용) | max_steps + max_tokens 이중 제한 |
MCP의 도구-모델 분리 아키텍처는 이러한 금융 규제 요건을 자연스럽게 충족합니다. 모델은 도구의 "사용자"일 뿐이고, 실제 데이터 접근과 비즈니스 로직은 MCP Server에 캡슐화되므로, 보안 감사 범위가 명확해집니다.
12. 정리
오늘 12화에서 다룬 내용을 압축하면:
- Qwen3 Function Calling은 OpenAI 호환 형식으로 동작하며, vLLM의
--enable-auto-tool-choice --tool-call-parser hermes로 활성화. 프롬프트 + Pydantic 검증 + 재시도 피드백의 3중 전략으로 실패율을 0.5% 미만으로 억제. - MCP로 사내 시스템을 표준화된 도구로 래핑하면, PII 마스킹·감사 로그·권한 격리가 서버 측에서 일관되게 처리됨. 에이전트(모델)는 도구의 "소비자"에 불과하므로 보안 경계가 명확.
- 오케스트레이션은 시나리오에 따라 ReAct(기본) → Plan-Execute(다단계) → 멀티 에이전트(다도메인) → LangGraph(복잡 워크플로) 순으로 확장. 80%의 시나리오는 ReAct로 충분.
- Qwen3-VL GUI 에이전트는 API 없는 레거시 시스템 자동화에 유용하나, 반자동(사람 승인 포함) 모드 권장.
모델이 도구를 갖추고 행동할 수 있게 되면, 다음 질문은 자연스럽게 "이 행동이 안전한가?"입니다.
내일 13화 예고: 입출력 가드레일, 환각 억제(grounding·인용), PII 마스킹, 휴먼인더루프, 그리고 오프라인/온라인 평가 — 에이전트를 프로덕션에 내보내기 전 반드시 갖춰야 할 안전·관측 레이어를 다룹니다.
📚 시리즈: 온프레미스 AI Assistant 아키텍처 — Qwen3·Qwen3-VL 14일 설계 (총 14화 중 12화)
◀ 이전 11화 (다음 차수는 아직 게시되지 않았습니다)
◀ 이전 11화 (다음 차수는 아직 게시되지 않았습니다)
참고 자료
- Qwen Function Calling 공식 문서 — Qwen 모델의 도구 호출 설정 및 사용법을 설명하는 공식 가이드
- Model Context Protocol 공식 사이트 — MCP 표준 프로토콜의 개념, 아키텍처, 구현 가이드를 제공하는 공식 문서
[…] 온프레미스 AI Assistant 아키텍처 — Qwen3·Qwen3-VL 14일 설계 (총 14화 중 13화)◀ 이전 12화 (다음 차수는 아직 게시되지 […]