CLOVA Studio 운영자6

들어가며 지난 쿡북 AI 어시스턴트 제작 레시피: Function calling 활용하기에서는 단순한 가상의 파이썬 함수를 사용해, 주식 주문을 흉내 내는 가벼운 시뮬레이션을 구현해보았는데요. 이번에는 그 흐름을 한층 더 확장해 보려 합니다. FastAPI로 주식 거래 API를 만들고, 이를 FastMCP를 통해 MCP 도구로 래핑하여 LLM이 진짜 도구를 호출하는 사례를 구현합니다. MCP(Model Context Protocol)는 LLM 및 외부 도구·서비스 간의 연결 방식을 표준화한 개방형 프로토콜로, 쉽게 말해 다양한 기능을 가진 도구들을 LLM이 일관된 방식으로 사용할 수 있게 해주는 어댑터 역할을 합니다. 흔히 'AI를 위한 USB-C'에 비유되며, 검색, 데이터베이스, API 등 어떤 도구든 동일한 포맷으로 연결할 수 있게 해줍니다. MCP는 JSON-RPC 2.0이라는 경량 메시지 포맷을 기반으로 통신하는데요. JSON은 {"key": "value"}처럼 사람이 읽을 수 있는 직관적인 구조이고, RPC(Remote Procedure Call)는 원격 서버에 기능 실행을 요청하는 방식입니다. REST API보다 간결하고 기능 중심적인 요청/응답 구조를 제공하는 것이 특징입니다. MCP는 크게 두 가지 기본 명령어로 구성됩니다. tools/list: MCP 서버에 등록된 모든 도구의 메타데이터(이름, 설명, 입력값 스키마 등)를 조회합니다. 이 정보는 LLM이 사용할 수 있도록 변환되어, 모델에게 '어떤 도구들이 어떤 방식으로 호출 가능한지'를 알려주는 역할을 합니다. tools/call: LLM이 생성한 toolCall 요청(도구 이름과 입력값)을 실제로 실행하고, 결과를 받아오는 명령입니다. 일반적인 실행 흐름은 다음과 같습니다. 클라이언트가 MCP 서버에 tools/list 요청을 보내면, MCP 서버는 JSON 포맷의 도구 목록을 반환합니다. 이 메타데이터는 LLM에게 입력으로 전달되어, 모델이 사용자 질문을 분석하고 적절한 toolCall 요청을 구성하도록 돕습니다. 이후 클라이언트는 해당 toolCall을 tools/call 형식으로 변환해 서버에 요청하고, 실행 결과를 받아 LLM에게 다시 전달합니다. 마지막으로 모델은 이 실행 결과를 반영한 자연어 응답을 생성해 사용자에게 전달합니다. 본 쿡북에서 사용하는 FastMCP는 Python 기반의 MCP 구현 라이브러리로, 기존 REST API를 손쉽게 MCP 도구로 전환할 수 있게 해줍니다. FastMCP.from_fastapi()를 통해 FastAPI로 구성한 REST API를 MCP 도구로 래핑할 수 있으며, 별도의 REST API가 없는 경우에는 @mcp.tool 데코레이터만 붙여도 단일 함수를 도구로 노출할 수 있습니다. 두 방식 모두 JSON-RPC 요청을 자동으로 파싱하고, 스키마를 자동으로 생성해 주는 등 구현 부담을 줄여준다는 장점이 있습니다. 본 쿡북에서는 MCP 서버의 tools/list 및 tools/call 응답을 Function calling 스키마로 매핑하여 모델 입력에 포함시키는 방식을 사용합니다. 즉, 도구 메타데이터를 모델이 이해하는 함수 정의로 재구성하는 변환 로직을 클라이언트에 구현했습니다. 참고로, OpenAI 호환 API를 활용하는 경우에는 이러한 매핑 없이도 클라이언트에 MCP 서버를 곧바로 연동할 수 있습니다. 시나리오 소개 FastAPI로 구성된 주식 거래 API를 FastMCP로 감싸 MCP 도구로 등록하고, LLM이 자연어 요청을 분석해 적절한 도구를 호출하는 워크플로우를 구성합니다. 사용자는 아래와 같은 질문을 자연스럽게 입력할 수 있습니다. 네이버 지금 얼마야? 네이버 1주 매수해줘 내 계좌 잔고 조회해줘. 비번은 **** 7월 한달 거래 내역 보여줘 주요 기능 요약 주가 조회: 특정 종목의 실시간 주가를 확인할 수 있습니다. 가격은 FinanceDataReader에서 가져온 실시간 정보를 사용합니다. 주식 주문: 사용자가 입력한 종목 코드에 대해 시장가(또는 전일 종가) 기준으로 거래를 실행합니다. 가격은 FinanceDataReader에서 가져온 실시간 정보를 사용합니다. 잔고 확인: 현재 남은 현금과 각 종목의 보유 수량 및 평균 단가를 반환합니다. 비밀번호가 필요하며, 예제의 비밀번호는 '1234'로 고정되어 있습니다. 거래 내역 조회: 기간을 지정하면 해당 기간의 매수/매도 내역을 필터링해 반환합니다. 날짜 형식은 ISO 표준(YYYY-MM-DD)을 따릅니다. 실행 흐름 시나리오 구현 본 예제는 시뮬레이션 목적의 데모 코드이며, 주식 매수 및 매도 시 평균 단가 및 손익 계산은 단순화되어 있습니다. 실제 서비스에 적용할 경우 정확한 회계 기준과 거래 규칙에 따라 로직을 재설계해야 합니다. 1. 설치 패키지 시나리오 구현을 위해서 먼저 다음의 파이썬 패키지를 설치합니다. FastAPI와 Uvicorn은 REST API 서버를 구동하기 위한 기본 조합이고, FastMCP는 MCP 도구 스펙 등록을 위해 필요합니다. 금융 데이터가 필요한 예제이므로 finance-datareader도 함께 설치합니다. pip install fastapi uvicorn fastmcp finance-datareader 만약 더 빠른 설치 속도를 원한다면 다음과 같이 uv를 설치합니다. curl -LsSf https://astral.sh/uv/install.sh | sh uv add fastmcp 2. 파일 구성 세 개의 주요 파일로 구성되며, LLM ↔ MCP 서버 ↔ 외부 데이터 소스 간 연동 흐름을 구현합니다. cookbook/ ├── my_client.py # 사용자 입력을 받아 LLM과 MCP 서버를 연동하는 실행 클라이언트입니다. ├── my_server.py # REST와 MCP 서버를 동시에 띄우는 실행용 서버 스크립트입니다. └── stock_api.py # FastAPI로 구현된 REST API로, 주식 주문 및 조회 기능을 제공합니다. my_server.py에 의해 MCP 도구로 래핑되어 활용됩니다. 3. 파일 생성 3.1 stock_api.py FastAPI로 구현된 REST API로, 주식 주문 및 조회 기능을 제공합니다. 현금 잔고, 보유 종목, 거래 내역을 모두 메모리 상태로 유지하며, 다음 네 가지 엔드포인트를 제공합니다. POST /buy: 종목 매수 POST /sell: 종목 매도 GET /balance: 현금 및 보유 종목 조회 GET/ trades: 거래 내역 조회 매수 또는 매도 요청이 들어오면 FinanceDataReader를 통해 실시간 시세를 조회한 뒤, 잔고를 갱신하고 체결 내역을 기록합니다. 튜토리얼 목적의 예제이므로 별도의 데이터베이스 없이 파이썬 전역 변수를 통해 상태를 유지합니다. import FinanceDataReader as fdr from fastapi import FastAPI, HTTPException, Query, Header, Depends from pydantic import BaseModel, Field from datetime import date, datetime from typing import Optional, Dict, List from typing import Any app = FastAPI(title="Stock Trading API", version="1.0.0") # 전역 상태 정의 cash_balance: int = 10_000_000 # 초기 현금 1천만 원 portfolio: Dict[str, Dict[str, int | str]] = {} # 종목별 보유 수량 및 평균 단가 trade_history: List[Dict[str, int | str]] = [] # 거래 내역 # 간단한 비밀번호 설정 ACCOUNT_PASSWORD = "1234" class PortfolioItem(BaseModel): """보유 종목 정보""" qty: int = Field(..., description="보유 수량") name: Optional[str] = Field(None, description="종목명") avg_price: int = Field(..., description="평균 단가") class TradeRequest(BaseModel): """매수/매도""" ticker: str = Field(..., description="종목 코드 (예: 035420)") qty: int = Field(..., gt=0, description="매수 또는 매도할 수량") class BalanceResponse(BaseModel): """잔고 조회""" available_cash: int = Field(..., description="현금 잔고(원)") portfolio: Dict[str, Any] = Field(..., description="종목별 보유 내역") class TradeHistoryItem(BaseModel): """거래 내역""" type: str = Field(..., description="거래 종류 (buy 또는 sell)") name: Optional[str] = Field(None, description="종목명") ticker: str = Field(..., description="종목 코드") qty: int = Field(..., description="거래 수량") price: int = Field(..., description="거래 체결 가격") avg_price: Optional[int] = Field(None, description="거래 후 평균 단가") datetime: str = Field(..., description="거래 시각 (YYYY-MM-DD HH:MM:SS)") def get_market_price(ticker: str) -> int: """주어진 종목 코드의 가장 최근 종가를 조회합니다. Args: ticker: 종목 코드 Returns: int: 최근 종가 Raises: HTTPException: 종목 데이터가 없는 경우 """ df = fdr.DataReader(ticker) if df.empty: raise HTTPException(status_code=404, detail=f"종목 {ticker}에 대한 시장 데이터를 찾을 수 없습니다.") return int(df['Close'].iloc[-1]) def get_corp_name(ticker: str) -> str: """종목 코드를 종목명으로 변환합니다. 못 찾으면 그대로 코드 반환.""" krx = fdr.StockListing("KRX") match = krx.loc[krx['Code'] == ticker, 'Name'] return match.values[0] if not match.empty else ticker @app.post("/buy", summary="종목 매수", operation_id="buy_stock", response_model=dict) async def buy_stock(trade: TradeRequest): """주어진 종목을 지정한 수량만큼 매수합니다. 요청 본문으로 종목 코드와 수량을 받으며, 현재 잔고가 부족하면 400 오류를 반환합니다. """ price = get_market_price(trade.ticker) name = get_corp_name(trade.ticker) cost = trade.qty * price # 잔고 확인 및 업데이트 global cash_balance global portfolio if cost > cash_balance: raise HTTPException(status_code=400, detail=f"잔고가 부족합니다. 현재 잔고는 {cash_balance:,}원이며, 총 {cost:,}원이 필요합니다.") cash_balance -= int(cost) existing = portfolio.get(trade.ticker, {"qty": 0, "avg_price": 0.0, "name": name}) total_qty = existing["qty"] + trade.qty avg_price = ((existing["qty"] * existing["avg_price"]) + cost) / total_qty portfolio[trade.ticker] = { "qty": total_qty, "name": name, "avg_price": int(round(avg_price, 2)) } trade_history.append({ "type": "buy", "name": name, "ticker": trade.ticker, "qty": trade.qty, "price": int(round(price, 2)), "avg_price": int(round(avg_price, 2)), "datetime": datetime.now().strftime("%Y-%m-%d %H:%M:%S") }) current_cash = cash_balance return { "message": f"{name} {trade.qty}주 매수 완료 (시장가 {round(price, 2)}원)", "available_cash": current_cash } @app.post("/sell", summary="종목 매도", operation_id="sell_stock", response_model=dict) async def sell_stock(trade: TradeRequest): """보유 종목을 지정한 수량만큼 매도합니다. 보유 수량이 부족하면 400 오류를 반환합니다. 매도 후 잔여 수량이 0이면 포트폴리오에서 삭제합니다. """ global cash_balance global portfolio if trade.ticker not in portfolio or portfolio[trade.ticker]["qty"] < trade.qty: raise HTTPException(status_code=400, detail=f"보유한 수량이 부족합니다. 현재 보유: {portfolio.get(trade.ticker, {}).get('qty', 0)}주, 요청 수량: {trade.qty:,}주") price = get_market_price(trade.ticker) name = get_corp_name(trade.ticker) revenue = trade.qty * price current_qty = portfolio[trade.ticker]["qty"] current_avg_price = portfolio[trade.ticker]["avg_price"] new_qty = current_qty - trade.qty cash_balance += int(revenue) if new_qty == 0: del portfolio[trade.ticker] else: portfolio[trade.ticker]["qty"] = new_qty portfolio[trade.ticker]["avg_price"] = current_avg_price trade_history.append({ "type": "sell", "name": name, "ticker": trade.ticker, "qty": trade.qty, "price": int(round(price, 2)), "avg_price": int(round(current_avg_price, 2)), "datetime": datetime.now().strftime("%Y-%m-%d %H:%M:%S") }) current_cash = cash_balance return { "message": f"{name} {trade.qty}주 매도 완료 (시장가 {round(price, 2)}원)", "available_cash": current_cash } @app.get("/balance", summary="잔고 조회", operation_id="get_balance", response_model=BalanceResponse) async def get_balance(password: str = Header(..., alias="X-Account-Password")): """현재 보유 현금과 포트폴리오를 반환합니다. 요청 시 HTTP 헤더의 `X-Account-Password` 값을 통해 비밀번호를 전달받습니다. """ if password != ACCOUNT_PASSWORD: raise HTTPException(status_code=401, detail="잘못된 비밀번호입니다.") typed_portfolio = {ticker: PortfolioItem(**data) for ticker, data in portfolio.items()} return { "available_cash": cash_balance, "portfolio": typed_portfolio } @app.get("/trades", summary="거래 내역 조회", operation_id="get_trade_history", response_model=List[TradeHistoryItem]) async def get_trade_history( start_date: Optional[date] = Query(None, description="조회 시작일 (예: 2025-07-01)"), end_date: Optional[date] = Query(None, description="조회 종료일 (예: 2025-07-28)") ): """지정 기간 동안의 거래 내역을 반환합니다.""" if start_date and end_date and start_date > end_date: raise HTTPException(status_code=400, detail="start_date는 end_date보다 이후일 수 없습니다.") filtered: List[Dict] = [] for trade in trade_history: trade_dt = datetime.strptime(trade["datetime"], "%Y-%m-%d %H:%M:%S") if start_date and trade_dt.date() < start_date: continue if end_date and trade_dt.date() > end_date: continue filtered.append(trade) return filtered @app.get("/", summary="서비스 안내") async def root() -> Dict[str, str]: return {"message": "Welcome to the Stock Trading API"} 3.2 my_server.py 앞서 정의한 주식 거래 API를 MCP 도구로 래핑해 한 번에 서비스하도록 구성한 스크립트입니다. FastMCP.from_fastapi()로 REST API를(주식 거래 API)을 MCP 도구 세트로 변환합니다. mcp.http_app()을 통해 /mcp/ 경로에 JSON-RPC 기반 MCP 서버를 마운트합니다. 동시에, 원본 REST API는 /api 경로에 그대로 유지해, REST와 MCP를 병행 노출합니다. 결과적으로 uvicorn으로 my_server:app만 띄우면, REST(/api)와 MCP(/mcp)가 동시에 구동됩니다. REST에 없는 기능도 별도로 등록 가능하며, 예시로 @mcp.tool() 데코레이터를 통해 get_price라는 도구를 추가 구현하였습니다. from fastapi import FastAPI from fastmcp import FastMCP from stock_api import app as stock_api_app def create_app() -> FastAPI: instructions = ( "이 MCP 서버는 주식 매수/매도, 잔고 조회, 거래 내역 조회, 시세 조회 기능을 제공합니다." ) # 1) 기존 FastAPI의 API 전체를 MCP 도구 세트로 래핑하고 MCP 서버 객체를 생성합니다. mcp = FastMCP.from_fastapi( stock_api_app, name="Stock Trading MCP", instructions=instructions, ) # 2) FastAPI API에 정의되지 않은 이외 도구를 @mcp.tool로 추가 등록합니다.(즉 API가 아닌 파이썬 개별 함수) # 즉 1)에서 생성한 서버에 도구 추가 @mcp.tool( name="get_price", description="특정 종목의 실시간 주가 또는 최근 종가를 반환합니다.", ) async def get_price(ticker: str) -> dict: """FinanceDataReader로 가장 최근 종가를 가져와 반환""" import FinanceDataReader as fdr df = fdr.DataReader(ticker) latest = df.iloc[-1] return { "ticker": ticker, "date": latest.name.strftime("%Y-%m-%d"), "close": int(latest["Close"]), } # 3) MCP JSON‑RPC 서브 앱 생성 mcp_app = mcp.http_app( path="/", transport="http", stateless_http=True, json_response=True, ) # 4) 루트 FastAPI에 REST와 MCP를 마운트 root_app = FastAPI( title="Stock Trading Service with MCP", lifespan=mcp_app.lifespan, ) root_app.mount("/api", stock_api_app) root_app.mount("/mcp", mcp_app) @root_app.get("/") async def root() -> dict: return { "message": "Stock Trading Service with MCP", "api": "/api", "mcp": "/mcp", } return root_app app = create_app() if __name__ == "__main__": import uvicorn uvicorn.run("my_server:app", host="0.0.0.0", port=8888, reload=True) 3.3 my_client.py 사용자로부터 자연어 입력을 받아 LLM에 전달하고, Function calling 기반 응답을 분석하여 필요한 MCP 도구를 자동으로 호출합니다. 이를 위해 mcp_client()를 통해 MCP 서버와 통신하기 위한 클라이언트 객체를 생성하고, 이 객체로 tools/list 및 tools/call 명령을 수행합니다. load_tools(): MCP 서버에서 등록된 도구 목록을 조회하고, Chat Completions v3 API 요청 포맷에 맞게 변환 call_llm(): Chat Completions v3 API를 호출해 사용자 질문에 대한 응답을 생성 call_mcp_tool(): toolCall 응답을 기반으로 MCP 서버의 도구를 실행 아래 코드에서 CLOVA Studio API 호출을 위해 본인이 발급 받은 API Key를 입력하세요. from fastmcp import Client from fastmcp.client.transports import StreamableHttpTransport import uuid import json import asyncio import requests from typing import Any, Dict, List # === CLOVA Studio 모델 엔드포인트 === MODEL_URL = "https://clovastudio.stream.ntruss.com/v3/chat-completions/HCX-005" MODEL_HEADERS = { "Authorization": "Bearer <YOUR_API_KEY>", # 실제 구현 시 Bearer 토큰을 환경 변수나 안전한 방법으로 관리하세요. "Content-Type": "application/json", } # === MCP 클라이언트 객체 정의 === transport = StreamableHttpTransport( url="http://localhost:8888/mcp/", headers={"X-Account-Password": "1234"} ) mcp_client = Client(transport) # === MCP에서 도구 스펙 받아와서 Function calling 포맷으로 변환 === async def load_tools(client: Client) -> List[Dict[str, Any]]: tools = await client.list_tools() tools_spec = [] for tool in tools: schema = tool.inputSchema or {} props = schema.get("properties", {}) if not props: continue tools_spec.append({ "type": "function", "function": { "name": tool.name, "description": tool.description or "", "parameters": { "type": schema.get("type", "object"), "properties": { k: { "type": p.get("type", "string"), "description": p.get("description", "") } for k, p in props.items() }, "required": schema.get("required", []) }, }, }) return tools_spec # === 모델 호출 === def call_llm(messages: List[Dict[str, Any]], tools_spec=None, tool_choice=None) -> Dict[str, Any]: data = {"messages": messages, "maxTokens": 1024, "seed": 0} if tools_spec: data["tools"] = tools_spec if tool_choice: data["toolChoice"] = tool_choice headers = MODEL_HEADERS | {"X-NCP-CLOVASTUDIO-REQUEST-ID": str(uuid.uuid4())} resp = requests.post(MODEL_URL, headers=headers, json=data) resp.raise_for_status() return resp.json() # === MCP 도구 실행 === async def call_mcp_tool(client: Client, name: str, args: Dict[str, Any]) -> Any: return await client.call_tool(name, args) # === main loop === async def main(): async with mcp_client as client: tools_spec = await load_tools(client) system_prompt = { "role": "system", "content": ( "당신은 사용자 주식 거래를 돕는 AI 어시스턴트입니다. " "매수·매도, 잔고 조회, 거래 내역 조회, 주가 조회를 처리하고 결과를 수치로 명확히 안내하세요. " "잔고·수량 부족 등 거래가 불가능하면 이유를 숫자와 함께 설명하세요." ), } while True: user_input = input("\n사용자 요청을 입력하세요: ") if user_input.lower() in {"exit", "quit", "종료"}: print("\n대화를 종료합니다.") break user_msg = {"role": "user", "content": user_input} first_resp = call_llm([system_prompt, user_msg], tools_spec=tools_spec, tool_choice="auto") if first_resp.get("status", {}).get("code") != "20000": print("\nLLM 호출 실패:", first_resp.get("status")) continue assistant_msg = first_resp["result"]["message"] tool_calls = assistant_msg.get("toolCalls", []) if not tool_calls: print("\n모델 답변:", assistant_msg.get("content", "")) continue tool_call = tool_calls[0] func_name = tool_call["function"]["name"] func_args = tool_call["function"]["arguments"] call_id = tool_call["id"] try: tool_result = await call_mcp_tool(client, func_name, func_args) except Exception as err: print("\nMCP 도구 실행 실패:", err) continue tool_response_prompt = { "role": "system", "content": ( "아래 tool 결과를 기반으로 간결하게 최종 답변을 작성하세요. " "'available_cash'는 현재 남은 현금 잔고, 'portfolio'는 종목별 보유 수량과 평균 단가입니다. " "수치는 단위와 함께 명확하게 표현하세요. (예: 3주, 1,000원)\n" "금액 해석 시 숫자의 자릿수를 기준으로 정확히 구분하세요." ), } second_resp = call_llm( [ tool_response_prompt, user_msg, {"role": "assistant", "content": "", "toolCalls": [tool_call]}, { "role": "tool", "toolCallId": call_id, "name": func_name, "content": json.dumps(tool_result.structured_content, ensure_ascii=False), }, ] ) if second_resp.get("status", {}).get("code") == "20000": print("\n모델 답변:", second_resp["result"]["message"]["content"]) else: print("\nLLM 호출 실패:", second_resp.get("status")) if __name__ == "__main__": asyncio.run(main()) 4. 실행 다음 명령어를 서로 다른 터미널에서 순서대로 실행합니다. uvicorn my_server:app --host 0.0.0.0 --port 8888 --reload 서버가 먼저 실행되어 있어야 클라이언트 호출을 정상적으로 수행할 수 있습니다. python3 my_client.py 클라이언트에 자연어로 요청을 입력하면 모델 응답 결과가 반환됩니다. 다음은 응답 예시입니다. 사용자 요청을 입력하세요: 안녕 모델 답변: 안녕하세요! 저는 CLOVA X입니다. 주식 거래에 필요한 정보를 제공해 드리며, 다음과 같은 기능을 수행할 수 있습니다. 1. 매수/매도: 원하는 종목과 수량을 입력하시면 해당 종목을 매수하거나 매도합니다. 2. 잔고 조회: 현재 보유 중인 종목 및 수량 등을 확인할 수 있습니다. 3. 거래 내역 조회: 과거에 진행한 거래 내역을 조회하여 상세 내용을 확인 가능합니다. 4. 주가 조회: 특정 종목의 현재 가격을 실시간으로 조회할 수 있습니다. 언제든지 필요하신 부분이 있으시면 말씀해 주세요. 최대한 도움이 되도록 노력하겠습니다. 사용자 요청을 입력하세요: 내 주식 계좌 잔고 조회해줘. 비번 1234 모델 답변: 현재 주식 계좌에 남아 있는 현금은 10,000,000원이며, 보유한 주식은 없습니다. 사용자 요청을 입력하세요: 네이버 5주 매수해줘 모델 답변: 네이버 5주를 시장가 234,500원에 매수하였습니다. 현재 계좌에는 8,827,500원이 남아 있습니다. 사용자 요청을 입력하세요: 네이버 주가 알려줘 모델 답변: 현재 네이버(035420)의 주가는 **235,000원**입니다. 사용자 요청을 입력하세요: 네이버 1개 더 매수 모델 답변: 네이버 주식 1주를 시장가 235,000원에 매수하였습니다. 현재 계좌에는 8,592,500원이 남아 있습니다. 사용자 요청을 입력하세요: 네이버 1개 팔아줘 모델 답변: 네이버 1주를 시장가 235,000원에 매도하였습니다. 현재 남아있는 현금은 8,827,500원입니다. 사용자 요청을 입력하세요: 내 계좌 잔고 다시 조회해줘. 비번 1234 모델 답변: 현재 계좌에는 **8,827,500원**이 남아있으며, **NAVER 5주**를 보유하고 있습니다. - 종목코드: 035420 - 이름: NAVER - 평균단가: 234,583원 사용자 요청을 입력하세요: 오늘 거래내역 표로 보여줘 모델 답변: 다음은 오늘(2025년 7월 31일) 거래 내역입니다. | 거래 유형 | 종목 이름 | 종목 코드 | 수량 | 가격 | 평균단가 | 거래 시간 | |------------|-----------|-----------|------|-------|----------|-----------| | 매수 | NAVER | 035420 | 5 | 234500 | 234500 | 15:16:21 | | 매수 | NAVER | 035420 | 1 | 235000 | 234583 | 15:16:43 | | 매도 | NAVER | 035420 | 1 | 235000 | 234583 | 15:17:30 | 총 2번의 매수와 1번의 매도가 있었습니다. NAVER 종목을 각각 5주와 1주씩 총 6주를 매수한 후, 1주를 매도하여 현재 NAVER 종목에 대한 보유 수량은 **5주**입니다. MCP 서버에 등록된 도구 목록 조회 다음 명령어를 실행하면 MCP 서버에 등록된 도구 목록을 조회할 수 있습니다. curl -L -X POST http://localhost:8888/mcp/ \ -H "Content-Type: application/json" \ -H "Accept: application/json, text/event-stream" \ -d '{"jsonrpc":"2.0","id":1,"method":"tools/list","params":{}}' { "jsonrpc": "2.0", "id": 1, "result": { "tools": [ { "name": "buy_stock", "description": "주어진 종목을 지정한 수량만큼 매수합니다.", "inputSchema": { "type": "object", "properties": { "ticker": { "type": "string", "title": "Ticker", "description": "종목 코드 (예: 035420)" }, "qty": { "type": "integer", "exclusiveMinimum": 0, "title": "Qty", "description": "매수 또는 매도할 수량" } }, "required": ["ticker", "qty"] }, "outputSchema": { "type": "object", "title": "Response Buy Stock", "additionalProperties": true } }, { "name": "sell_stock", "description": "보유 종목을 지정한 수량만큼 매도합니다.", "inputSchema": { "type": "object", "properties": { "ticker": { "type": "string", "title": "Ticker", "description": "종목 코드 (예: 035420)" }, "qty": { "type": "integer", "exclusiveMinimum": 0, "title": "Qty", "description": "매수 또는 매도할 수량" } }, "required": ["ticker", "qty"] }, "outputSchema": { "type": "object", "title": "Response Sell Stock", "additionalProperties": true } }, { "name": "get_balance", "description": "현재 보유 현금과 포트폴리오를 반환합니다.", "inputSchema": { "type": "object", "properties": { "X-Account-Password": { "type": "string", "title": "X-Account-Password" } }, "required": ["X-Account-Password"] }, "outputSchema": { "type": "object", "title": "BalanceResponse", "required": ["available_cash", "portfolio"], "properties": { "available_cash": { "type": "integer", "title": "Available Cash", "description": "현금 잔고(원)" }, "portfolio": { "type": "object", "title": "Portfolio", "description": "종목별 보유 내역", "additionalProperties": true } } } }, { "name": "get_trade_history", "description": "지정 기간 동안의 거래 내역을 반환합니다.", "inputSchema": { "type": "object", "properties": { "start_date": { "anyOf": [ { "type": "string", "format": "date" }, { "type": "null" } ], "title": "Start Date" }, "end_date": { "anyOf": [ { "type": "string", "format": "date" }, { "type": "null" } ], "title": "End Date" } } }, "outputSchema": { "type": "object", "required": ["result"], "properties": { "result": { "type": "array", "title": "Response Get Trade History", "items": { "$ref": "#/$defs/TradeHistoryItem" } } }, "$defs": { "TradeHistoryItem": { "type": "object", "title": "TradeHistoryItem", "description": "거래 내역 항목.", "required": [ "type", "ticker", "qty", "price", "datetime" ], "properties": { "type": { "type": "string", "description": "거래 종류 (buy 또는 sell)" }, "name": { "anyOf": [ { "type": "string" }, { "type": "null" } ], "title": "Name" }, "ticker": { "type": "string", "title": "Ticker" }, "qty": { "type": "integer", "title": "Qty" }, "price": { "type": "integer", "title": "Price" }, "avg_price": { "anyOf": [ { "type": "integer" }, { "type": "null" } ], "title": "Avg Price" }, "datetime": { "type": "string", "title": "Datetime", "description": "거래 시각 (YYYY-MM-DD HH:MM:SS)" } } } }, "x-fastmcp-wrap-result": true } }, { "name": "root", "description": "서비스 안내", "inputSchema": { "type": "object", "properties": {}, "required": [] }, "outputSchema": { "type": "object", "additionalProperties": { "type": "string" } } }, { "name": "get_price", "description": "특정 종목의 최신 종가를 반환합니다.", "inputSchema": { "type": "object", "properties": { "ticker": { "type": "string", "title": "Ticker" } }, "required": ["ticker"] }, "outputSchema": { "type": "object", "additionalProperties": true } } ] } } 마무리 이번 예제는 FastMCP와 Function calling을 결합해, 사용자의 자연어 입력을 기반으로 LLM이 적절한 도구를 자동 선택하고 실행하는 전체 흐름을 구현해 보았습니다. 지정가 매수, 사용자별 계정 관리, 대화 히스토리 저장 등 다양한 방향으로 확장도 가능하니, 나만의 유용한 애플리케이션으로 발전시켜 보세요!

@CHOI님 안녕하세요. 토큰 수 초과로 인해 발생한 오류는 아래 가이드를 참고하여 해결해 보실 수 있습니다. https://guide.ncloud-docs.com/docs/clovastudio-skill#api-spec-확장-기능 또한, 데이터 수집 시 토큰 수를 합산하는 방식에 대해서는 아래 가이드를 참고해 주세요. https://guide.ncloud-docs.com/docs/clovastudio-skilltrainer-faq#데이터-수집 그 외 추가 문의사항이 있으시면 언제든지 편하게 남겨 주세요. 감사합니다.

Postman Flows를 활용해 CLOVA Studio API와 외부 API를 시각적으로 연결하고, 실제 애플리케이션처럼 동작하는 워크플로우를 구현하는 방법을 소개합니다. 최근에는 Dify, n8n 등 GUI 기반의 노코드/로우코드 워크플로우 도구들이 등장하면서, 외부 서비스 연동, 조건 분기, LLM 활용 등 복합적인 작업을 시각적으로 구성할 수 있는 환경이 떠오르고 있습니다. 개발 경험이 없는 사용자도 손쉽게 자동화를 설계하고 실행할 수 있도록 돕는 것이 특징인데요, Postman Flows 역시 이러한 흐름에 맞춘 노코드 도구로, API 요청을 시각적으로 연결해 복잡한 자동화 시나리오를 쉽게 구성할 수 있습니다. 이번 예제에서는 여러 API 호출을 순차적으로 연결해 하나의 자동화 시나리오를 완성해 보겠습니다. 1. 시나리오 소개 이번 예제에서는 회의 기록을 요약해 공유하고, 회의 중 나온 일정 정보를 자동으로 캘린더에 등록하는 흐름을 구성합니다. 이를 위해 CLOVA Studio의 Chat Completions V3 API와 Google Drive, Slack, Google Calendar API를 불러오고, 각 단계를 Postman Flows 상에서 시각적으로 연결하여 자동화를 구현합니다. 전체 흐름은 다음과 같습니다. Google Drive에 저장된 회의록 파일을 불러옵니다. Chat Completions V3 API를 통해 회의록을 요약하고, 동시에 회의 중 언급된 일정 및 액션 아이템을 추출합니다. 요약 결과는 Slack 메시지로 전송합니다. 추출된 일정 정보는 Google Calendar에 등록합니다. 2. 사전 준비 본 예제를 시작하기 전에, Google Drive, Google Calendar, Slack API를 연동하기 위한 인증 정보 발급 및 권한 설정 방법을 안내합니다. 2.1 Google Drive 및 Google Calendar API 준비 Google Drive API와 Google Calendar API를 활용하기 위한 설정 방법입니다. 두 API 모두 설정 방식이 동일하므로, 하나의 절차로 함께 안내합니다. 다음 과정에 대한 자세한 설명은 Google Cloud 가이드를 참고하세요. 먼저 Google API를 사용하려면 Google 계정이 필요하며, Google Cloud Console에 로그인할 수 있어야 합니다. 프로젝트 생성 Google Cloud Console에 접속하여 상단의 [프로젝트 선택] 버튼을 클릭합니다. [새 프로젝트] 버튼을 클릭한 뒤, 필수 정보들을 입력하고 [만들기]를 클릭합니다. API 활성화 좌측 메뉴에서 API 및 서비스 > 라이브러리로 이동합니다. Google Drive API, Google Docs API, Google Calendar API를 하나씩 검색하여 각 페이지에서 [사용] 버튼을 클릭해 활성화합니다. 참고로 Google Docs API는 직접 호출하지 않더라도, Google Drive API를 통해 문서 파일을 불러오기 위해서 함께 활성화가 필요합니다. OAuth 동의 화면 구성 좌측 메뉴에서 API 및 서비스 > OAuth 동의 화면으로 이동합니다. [시작하기] 버튼을 클릭한 뒤, 앱 정보와 대상(반드시 외부로 설정), 연락처 정보를 입력하고 [만들기] 버튼을 클릭합니다. OAuth 클라이언트 ID 생성 클라이언트 메뉴로 이동하여 [+ 클라이언트 만들기] 버튼을 클릭합니다. 애플리케이션 유형은 '웹 애플리케이션'을 선택하고, 하단의 승인된 리디렉션 URI 영역에 'https://oauth.pstmn.io/v1/callback'를 입력한 뒤 [만들기] 버튼을 클릭합니다. 생성된 클라이언트 ID와 보안 비밀번호를 복사해 둡니다. 테스트 사용자 등록 대상 메뉴로 이동하여 [+ Add users] 버튼을 클릭하고, 사용자 이메일(구글 계정)을 추가합니다. 2.2 Slack API 준비 Slack API를 활용하기 위한 설정 방법을 안내합니다. 다음 과정에 대한 자세한 설명은 Slack API 가이드를 참고하세요. Slack API를 사용하려면 Slack 계정이 필요하며, 사전에 워크스페이스와 채널이 생성되어 있어야 합니다. 앱 생성 Slack API에 접속하여 [Create an App] 버튼을 클릭하고, 팝업이 나타나면 [From scratch]을 선택합니다. 필수 정보를 입력한 뒤 [Create App] 버튼을 클릭합니다. OAuth 권한 설정 생성한 앱을 클릭한 뒤, 좌측 메뉴에서 OAuth & Permissions로 이동합니다. Scopes 영역으로 내려가 User Token Scopes에서 [Add an OAuth Scope] 버튼을 클릭한 뒤, 'calls: write' 권한을 추가합니다. OAuth Tokens 영역에서 [Install to Workspace] 버튼을 클릭하고, 발급된 토큰을 복사해 둡니다. Redirect URLs 영역에서 [Add New Redirect URL] 버튼을 클릭하고, 'https://oauth.pstmn.io/v1/callback'를 입력한 뒤 [Add] 버튼, [Save URLs] 버튼을 차례대로 클릭합니다. 3. API 구성 이번 단계에서는 이후 Flow 생성을 위해 Postman에서 Google Drive, Google Calendar, Slack API 요청을 사전에 구성하는 작업을 안내합니다. 아래의 Postman Collection 파일에는 본 예제에 사용되는 API 정보가 포함되어 있습니다. 좌측의 Collections 탭을 선택한 후 [Import] 버튼을 클릭하여 업로드할 수 있습니다. [Cookbook] Postman_example_collection.json 3.1 및 3.2 단계에서 필요한 인증은 아래 공통 절차를 참고하여 구성해 주세요. 3.1 Google Drive API 파일 이름을 기준으로 파일 ID를 검색한 뒤, 해당 파일 ID를 이용해 파일 내용을 조회하는 흐름을 구성합니다. API 명세에 대한 자세한 설명은 Google Drive API 가이드에서 확인 할 수 있습니다. 3.1.1 파일 ID 조회 먼저, 파일 이름으로 파일 ID를 조회하는 요청을 구성합니다. Postman 화면의 좌측 메뉴에서 Collections을 클릭한 후, 상단의 [New] 버튼을 클릭하여 새로운 HTTP 요청을 생성합니다. 다음과 같이 요청 정보를 입력합니다: 3.1.2 파일 콘텐츠 조회 다음은 조회한 파일 ID를 기반으로 해당 문서의 본문 내용을 불러오는 요청입니다. Postman 화면의 좌측 메뉴에서 Collections을 클릭한 후, 상단의 [New] 버튼을 클릭하여 새로운 HTTP 요청을 생성합니다. 다음과 같이 요청 정보를 입력합니다: 3.2 Google Calendar API Google Calendar에 새 이벤트를 생성하는 요청을 구성합니다. calendarID는 Google Calendar 웹 화면에서 [설정 > 캘린더 통합] 메뉴를 통해 확인할 수 있습니다. API 명세에 대한 자세한 설명은 Google Calendar API 가이드에서 확인 할 수 있습니다. Postman 화면의 좌측 메뉴에서 Collections을 클릭한 후, 상단의 [New] 버튼을 클릭하여 새로운 HTTP 요청을 생성합니다. 다음과 같이 요청 정보를 입력합니다: 3.3 Slack 메시지 전송 API Slack 채널에 메시지를 전송하는 요청을 구성합니다. API 명세에 대한 자세한 설명은 Slack API 가이드(#chat.postMessage)에서 확인 할 수 있습니다. Postman 화면의 좌측 메뉴에서 Collections을 클릭한 후, 상단의 [New] 버튼을 클릭하여 새로운 HTTP 요청을 생성합니다. 다음과 같이 요청 정보를 입력합니다: 3.4 Chat Completions V3 API 회의록 요약을 생성하고 캘린더 일정을 추출하는 LLM 요청을 생성합니다. API 명세에 대한 자세한 설명은 CLOVA Studio API 가이드에서 확인 할 수 있습니다. Postman 화면의 좌측 메뉴에서 Collections을 클릭한 후, 상단의 [New] 버튼을 클릭하여 새로운 HTTP 요청을 생성합니다. 다음과 같이 요청 정보를 입력합니다: 4. Flow 구현 및 테스트 이제 각 API를 연동하여 전체 자동화 흐름을 구성합니다. Google Drive에서 회의 기록을 불러오고, Chat completions API로 주요 내용을 요약한 뒤, Function calling 기능을 통해 Slack에 메시지를 전송하고 Google Calendar에 일정을 등록하는 흐름을 단계별로 구현합니다. 4.1 Flow 생성 다음 절차에 따라 flow를 생성합니다. Postman을 실행한 후 좌측 메뉴에서 Flows를 클릭한 후, 상단의 [New] 버튼을 클릭합니다. 하단에 [+ Block] 버튼을 클릭하여 원하는 블록을 생성합니다. 본 쿡북 예제에서 사용되는 블록 유형은 다음과 같습니다: 마우스로 드래그하여 블록 간 연결을 구현합니다. 4.2 Flow 실행 화면을 따라서 구현하세요. 연결이 완료되면 우측 Scenarios 메뉴를 클릭한 뒤, [+ Create scenario 버튼]을 클릭하여 회의 기록이 저장된 Google Drive의 파일명을 입력하여 저장합니다. [Run] 버튼을 클릭하여 전체 워크플로우를 실행합니다. 정상적으로 실행되면, 마지막 Display 블록에 성공 문구가 노출됩니다. 4.2 실행 화면 완성된 flow 및 시나리오 테스트 결과입니다. 4.3 실행 결과 시나리오에 활용된 회의 기록 및 Slack 및 Google Calendar 자동 전송 결과입니다. 회의 기록은 임의로 생성한 가상의 데이터입니다. 5. 맺음말 이번 쿡북에서는 LLM과 외부 서비스를 연동해 회의록 요약과 일정 등록 시나리오를 구현해 보았습니다. Postman Flows와 CLOVA Studio API를 활용하면, 실제 업무에 필요한 자동화를 노코드로 손쉽게 만들 수 있다는 점을 확인하셨을 텐데요. 이와 같은 흐름은 다양한 워크플로우로 확장할 수 있습니다. 예를 들어, 라우터와 스킬셋, Chat completions API 등을 조합해 나만의 AI 에이전트를 만들거나, 여러 LLM API를 연결해 응답을 비교·평가하는 인터페이스를 구현할 수 있습니다. 또는 Function calling을 통해 사용자의 자연어 입력을 SQL로 변환한 뒤, 로그 데이터를 조회하고 결과를 차트로 시각화하는 데이터 분석 도구로 확장해 볼 수도 있습니다. 이번 예제를 바탕으로 여러분의 아이디어와 업무 흐름에 맞는 자동화 시나리오를 자유롭게 설계해 보시기 바랍니다!

들어가며 이 쿡북에서는 Streamlit을 활용해 스킬 트레이너의 스킬셋과 라우터, 그리고 Chat Completions API를 결합하여 간단하게 AI 에이전트를 구축하는 방법을 알아보겠습니다. 다음 가이드를 따라 지역 검색 AI 에이전트를 직접 만들어 보고, 자신만의 AI 에이전트를 제작해 보세요! 🚀 전체 구조 각 기능별로 파일을 나누어 관리합니다. 스킬셋, 라우터, 그리고 Chat Completions와 같은 API 기능들을 별도의 파일로 만들어 필요할 때마다 불러와 사용할 수 있도록 구성하고, main.py 파일에서 이 모든 파일을 연결해서 실행할 수 있습니다. project/ │ ├── config.py # API 인증 정보를 관리합니다. ├── router.py # 라우터 API를 호출하여 사용자의 요청을 분류 및 필터링 합니다. ├── skillset.py # 스킬셋 API를 호출하여 답변을 생성합니다. ├── chat_completions.py # Chat Completions API를 호출하여 답변을 생성합니다. ├── chat_utils.py # 공통적으로 사용되는 유틸리티 함수를 관리합니다. └── main.py # Streamlit을 활용한 메인 앱 실행 파일입니다. 에이전트 워크플로우 본 쿡북에서는 지역 검색 에이전트를 제작합니다. 지역 검색 에이전트를 구성하는 주요 요소에 대한 제작 가이드는 다음 링크를 참고해 주세요. 지역 검색 스킬셋의 라우터: https://guide.ncloud-docs.com/docs/clovastudio-router-usecase 지역 검색 스킬셋: https://guide.ncloud-docs.com/docs/clovastudio-skilltrainer-usecase Chat Completions: https://api.ncloud-docs.com/docs/clovastudio-chatcompletions 사용자가 입력한 내용은 먼저 라우터에서 검토됩니다. 라우터는 사용자의 요청을 분석하여 적합한 도구(스킬셋 또는 Chat Completions)를 선택합니다. 이 과정에서 안전하지 않거나 부적절한 요청은 필터링되어 고정 응답이 반환됩니다. 스킬셋이 호출되면, 해당 스킬셋 내에 정의된 스킬(API)을 통해 실시간으로 데이터를 가져와 답변을 생성합니다. 반면, Chat Completions이 호출되면, 설정된 시스템 프롬프트와 파라미터 값에 기반하여 LLM이 답변을 생성합니다. 환경 설정 1. 필요한 라이브러리 설치 프로젝트를 시작하기 전에 필요한 Python 라이브러리를 설치합니다. pip install streamlit requests 2. API 키 설정 config.py 파일에서 API 호출 경로 및 인증 정보를 관리합니다. 아래 코드의 값 부분에 연동하고자 하는 라우터 및 스킬셋의 경로와 발급 받은 본인의 API 인증 정보를 넣어주세요. API Key는 절대 소스 코드에 노출되지 않도록 주의하세요. 환경 변수나 별도의 설정 파일을 사용하여 관리하는 것을 추천합니다. class Config: # 지역 검색의 라우터 호출 경로 ROUTER_API = 'YOUR_API_URL' # 지역 검색 스킬셋 호출 경로 SKILLSET_API = 'YOUR_API_URL' # 지역 검색 스킬셋에 정의된 스킬(API)의 인증 정보 NAVER_LOCAL_CLIENT_ID = 'YOUR_NAVER_CLIENT_ID' NAVER_LOCAL_CLIENT_SECRET = 'YOUR_NAVER_CLIENT_SECRET' # Chat Completions 호출 경로 CHAT_COMPLETIONS_API = 'YOUR_API_URL' # CLOVA Studio API 인증 정보 API_KEY = 'YOUR_API_KEY' REQUEST_ID_ROUTER = 'YOUR_REQUEST_ID' REQUEST_ID_SKILLSET = 'YOUR_REQUEST_ID' REQUEST_ID_CHAT = 'YOUR_REQUEST_ID' 각 모듈 설명 1. 라우터 API 호출: router.py 라우터는 사용자의 입력 내용을 분석하여, 어떤 도구를 사용해야 할지 결정하고(도메인 분류) 부적절한 내용을 감지(필터링)합니다. import requests from config import Config def get_router(query, chat_history=None): url = Config.ROUTER_API headers = { 'Authorization': f'Bearer {Config.API_KEY}', 'X-NCP-CLOVASTUDIO-REQUEST-ID': f'{Config.REQUEST_ID_ROUTER}', 'Content-Type': 'application/json' } data = { 'query': query } if chat_history and len(chat_history) >= 3: # 직전 user 턴의 발화를 가져옵니다. filtered_chat_history = chat_history[-3] data['chatHistory'] = [filtered_chat_history] response = requests.post(url, headers=headers, json=data) return response.json() 2. 스킬셋 API 호출: skillset.py 지역 검색 스킬셋 API를 호출하여 실시간 검색 데이터를 기반으로 한 답변을 생성합니다. import requests from config import Config def get_skillset(query, chat_history=None): url = Config.SKILLSET_API headers = { 'Authorization': f'Bearer {Config.API_KEY}', 'X-NCP-CLOVASTUDIO-REQUEST-ID': f'{Config.REQUEST_ID_SKILLSET}', 'Content-Type': 'application/json', } data = { 'query': query, 'requestOverride': { 'baseOperation': { 'header': { 'X-Naver-Client-Id': Config.NAVER_LOCAL_CLIENT_ID, 'X-Naver-Client-Secret': Config.NAVER_LOCAL_CLIENT_SECRET } } } } if chat_history: # 직전 user 턴의 발화 및 assistant 턴의 답변을 가져옵니다. filtered_chat_history = chat_history[-3:-1] data['chatHistory'] = filtered_chat_history response = requests.post(url, headers=headers, json=data) return response.json() 3. Chat Completions API 호출: chat_completions.py 시스템 프롬프트 및 파라미터 값과 함께 Chat Completions API를 호출하여 유연한 대화 흐름을 이끌어 줄 수 있는 답변을 생성합니다. import requests from config import Config def get_chat_response(query, chat_history=None): url = Config.CHAT_COMPLETIONS_API headers = { 'Authorization': f'Bearer {Config.API_KEY}', 'X-NCP-CLOVASTUDIO-REQUEST-ID': f'{Config.REQUEST_ID_CHAT}', 'Content-Type': 'application/json', } system_prompt = """[1. 지시문]\n당신에 대해 소개할 때는 [1-1. 아이덴티티]의 내용을 기반으로 말하세요.\n만약 당신에게 \"어떻게 질문하면 돼?\", \"어떤식으로 물어보면 돼?\", \"어떻게 질문하면 되는걸까요?\", \"사용방법 알려줘', \"사용방법 안내해 주세요\", \"사용방법을 알려줄 수 있을까요?\", \"사용방법 자세하게 알려줘\" 등과 같이 질문 방법에 대해 문의할 경우, 당신은 반드시 아래의 [1-2. 핵심 기능]과 [1-3. 예시 질문]에 관한 내용만을 응답해야 합니다. 반드시 아래에 제공된 정보만을 사용해야 하며, 주어지지 않은 정보를 임의로 생성하거나 추가하면 절대로 안 됩니다. \n\n[1-1. 아이덴티티]\n- 당신은 **실시간 장소 탐색 AI 에이전트**입니다.\n- 당신을 만든 곳은 Skill팀입니다. \b\n- 스킬셋 및 라우터 기능을 결합한 데모로 당신이 제작되었습니다. \n- 당신은 특정 지역의 맛집, 카페, 명소 등을 추천해 줄 수 있습니다.\n\n[1-2. 핵심 기능]\n지역 검색 : 사용자가 지역과 키워드를 바탕으로 질문하면(예: \"[특정 지역] 근처 맛집 추천해줘\") 네이버 지역 서비스에 등록된 정보를 기반으로 다양한 장소를 추천합니다.\n2) 유연한 대화 : 사용자의 질문 의도를 파악하고 다양한 표현으로 질문해도 정확하게 이해합니다.\n\n[1-3. 예시 질문]\n1)[지역]+[키워드] 추천해줘 (예: \"[특정 지역] 맛집 추천해줘\")\n\n[2. 지시문]\n만약 아래의 [2-1. 제한 사항]에 관련한 요청이 들어오면 답변이 불가능한 이유를 충분히 설명하고, 반드시 [1-2. 핵심 기능]과 [2-2. 예시]을 참고하여 적극적으로 대체 질문을 제안하거나 유도하세요.\n\n[2-1. 제한 사항]\n- 장소 탐색과 관련이 없는 실시간 정보 : 날씨, 주가, 시세 등의 정보에는 답변할 수 없습니다. \n- 지나치게 주관적인 질문 : 개인적인 취향에 대한 질문에는 답변하기 어렵습니다.\n\n[2-2. 예시]\n- 죄송합니다, 해당 정보는 제공할 수 없습니다. 대신 \"서울에서 가볼 만한 장소를 추천해줘\"와 같은 질문을 해 보시는 것도 좋을 것 같아요!\n- 대신 다른 정보를 도와드릴 수 있어요! 예를 들어, \"정자역 근처 맛집을 추천해줘\"와 같은 질문을 해 보시는 건 어떨까요?\n- 저는 실시간 장소 탐색 AI 에이전트이기 때문에 해당 정보는 제공할 수 없지만, 다른 정보가 궁금하시면 말씀해 주세요! 예를 들어, \”\강남역 카페 추천\”과 같은 질문은 어떠세요?""" messages = [{'role': 'system', 'content': system_prompt}] if chat_history: messages.extend(chat_history[-3:]) else: messages.append({'role': 'user', 'content': query}) data = { 'messages': messages, "maxTokens": 512, "seed": 0, "temperature": 0.4, "topP": 0.4, "topK": 0, "repeatPenalty": 5.0 } response = requests.post(url, headers=headers, json=data) return response.json() 4. 유틸리티 함수: chat_utils.py 공통적으로 사용되는 유틸리티 함수를 정리합니다. 본 쿡북에서는 응답 텍스트를 단어 단위로 스트리밍하여 출력하는 함수를 정의하였습니다. import time def streaming_data(text): for word in text.split(" "): yield word + " " time.sleep(0.05) 5. Streamlit 메인 앱 실행: main.py Streamlit을 활용한 메인 앱 실행 모듈입니다. 사용자 입력이 들어오면, 반드시 라우터를 거쳐 처리 방식을 결정하게 되고, 결과에 따라 스킬셋 답변이나 Chat Completions 답변, 고정 응답이 반환됩니다. display_response 함수를 통해 UI 상에 답변을 노출하고 대화 세션을 업데이트 합니다. import streamlit as st from router import get_router from chat_utils import streaming_data from chat_completions import get_chat_response from skillset import get_skillset def initialize_chat_session(): """에이전트 세션 초기화""" if 'messages' not in st.session_state: st.session_state.messages = [ { 'role': 'assistant', 'content': '안녕하세요. 장소 탐색 AI Agent입니다.😃 \n\n어떤 곳을 찾고 계신가요? 궁금하신 장소 정보가 있다면 언제든지 말씀해 주세요.' } ] def render_initial_messages(): """메시지 렌더링""" with st.chat_message(st.session_state.messages[0]['role']): st.write(st.session_state.messages[0]['content']) for message in st.session_state.messages[1:]: with st.chat_message(message['role']): st.write(message['content']) def display_response(final_answer): """응답 표시 및 세션 상태 업데이트""" with st.chat_message('assistant'): st.write_stream(streaming_data(final_answer)) st.session_state.messages.append({'role': 'assistant', 'content': final_answer}) def process_router(query, chat_history): """라우터 호출""" with st.status("라우터 적용 중...", expanded=True) as router_status: process_view = st.empty() process_view.write("라우터 적용 중입니다.") router_result = get_router(query, chat_history) domain = router_result.get('result', {}).get('domain', {}).get('result', '') blocked_content = router_result.get('result', {}).get('blockedContent', {}).get('result', []) safety = router_result.get('result', {}).get('safety', {}).get('result', []) return domain, blocked_content, safety, router_status, process_view def generate_skillset_response(query, chat_history): """지역 검색 스킬셋 응답 생성""" with st.status("답변 생성 중...", expanded=True) as answer_status: process_view = st.empty() process_view.write("API를 호출하고 답변을 생성하는 중입니다. 잠시만 기다려주세요.") result = get_skillset(query, chat_history) final_answer = result.get('result', {}).get('finalAnswer', '답변을 생성할 수 없습니다.') process_view.write("답변 생성이 완료되었습니다.") answer_status.update(label="답변 생성 완료", state="complete", expanded=False) return final_answer def generate_chat_response(query, chat_history): """chat_completions 응답 생성""" with st.status("답변 생성 중...", expanded=True) as answer_status: process_view = st.empty() process_view.write("요청하신 내용에 대한 답변을 생성 중입니다. 잠시만 기다려주세요.") result = get_chat_response(query, chat_history) final_answer = result.get('result', {}).get('message', {}).get('content', '답변을 생성할 수 없습니다.') process_view.write("답변 생성이 완료되었습니다.") answer_status.update(label="답변 생성 완료", state="complete", expanded=False) return final_answer def generate_filtered_response(filter_type): """고정 응답""" if filter_type == 'content': return ( "**콘텐츠 필터 규정**에 따라, 해당 질문에는 답변을 제공해 드리기 어려운 점 양해 부탁드려요.\n\n" "혹시 이렇게 질문해 보시는 건 어떠실까요? :)\n" "- 경기도 가을 단풍 명소 추천해 주세요.\n" "- 제주도 애월 맛집과 카페" ) else: # safety filter return ( '**안전 관련 규정**에 따라, 해당 질문에는 답변을 제공해 드리기 어려운 점 양해 부탁드려요.\n\n' '혹시 이렇게 질문해 보시는 건 어떠실까요?\n' '- 부산에서 인기 있는 맛집 찾아줄래?\n' '- 서울 분위기 좋은 카페 추천\n' '- 티엔미미 인기 메뉴 알려주세요\n\n' '언제나 좋은 정보로 도움 드리고자 합니다. 필요하신 내용이 있으시면 편하게 말씀해 주세요! 😊' ) def main(): st.set_page_config(page_title="장소 탐색 에이전트") st.title('장소 탐색 에이전트', anchor=False) st.write(' ') initialize_chat_session() render_initial_messages() if query := st.chat_input('질문을 입력하세요.'): with st.chat_message('user'): st.write(query) st.session_state.messages.append({'role': 'user', 'content': query}) chat_history = [{'role': msg['role'], 'content': msg['content']} for msg in st.session_state.messages] domain, blocked_content, safety, router_status, process_view = process_router(query, chat_history) router_status.update(label="라우터 적용 중...", state="running", expanded=True) if domain == "지역 검색": if not blocked_content and not safety: process_view.write("지역 검색 스킬셋으로 처리 가능합니다.") router_status.update(label="라우터 적용 완료", state="complete", expanded=False) final_answer = generate_skillset_response(query, chat_history) display_response(final_answer) elif blocked_content and not safety: process_view.write("스킬셋 사용이 불가능합니다. (이유 : 콘텐츠 필터)") router_status.update(label="라우터 적용 완료", state="complete", expanded=False) final_answer = generate_filtered_response('content') display_response(final_answer) else: process_view.write("스킬셋 사용이 불가능합니다. (이유 : 세이프티 필터)") router_status.update(label="라우터 적용 완료", state="complete", expanded=False) final_answer = generate_filtered_response('safety') display_response(final_answer) else: process_view.write("스킬셋과 관련 없는 요청입니다.") router_status.update(label="라우터 적용 완료", state="complete", expanded=False) final_answer = generate_chat_response(query, chat_history) display_response(final_answer) if __name__ == '__main__': main() main.py 상세 설명 main.py 파일의 구조를 자세히 살펴보면 에이전트가 어떻게 작동하는지 깊이 이해할 수 있고, 필요에 따라 해당 파일을 통해 에이전트의 기능을 추가로 커스텀 할 수 있습니다. 에이전트 실행하기 프로젝트 root 경로에서 터미널로 다음 명령어를 실행합니다. streamlit run main.py 실행 화면 예시 사용 시나리오 예시 오류 케이스 처리 팁 실제 서비스에서는 예상치 못한 문제들이 발생할 수 있습니다. 예를 들어, API 요청 시에 오류가 발생하거나 네트워크가 불안정할 수 있습니다. 이러한 오류 상황을 적절히 처리하면 에이전트의 안정성을 높이고 사용자 경험을 향상시킬 수 있으니, 개발 단계에서부터 다양한 오류 상황을 고려하여 코드를 작성하는 것이 좋습니다. 대표적으로 발생할 수 있는 몇가지 오류 상황과 처리 방법에 대한 코드 예시를 소개합니다. 1. API 응답 실패 API 호출 과정 중 발생할 수 있는 오류를 다음과 같이 간단하고 직관적으로 처리할 수 있습니다. 간소화된 메시지로 사용자에게 결과를 안내하고, 답변 생성 성공 또는 실패 상태를 UI에 명확히 표시해 줍니다. def get_skillset(query, chat_history=None): # ... (기존 코드) response = requests.post(url, headers=headers, json=data) if response.status_code == 200: return response.json() elif response.status_code == 400: return { 'result': '스킬 클라이언트 오류 발생', 'detail': response.json().get('status').get('code') } elif response.status_code == 500: return { 'result': '스킬 서버 오류 발생', 'detail': response.json().get('status').get('code') } else: return { 'result': '알 수 없는 오류 발생', 'detail': response.status_code } def generate_skillset_response(query, chat_history): with st.status("답변 생성 중...", expanded=True) as answer_status: # ... (기존 코드) result = get_skillset(query, chat_history) final_answer = result.get('result', {}).get('finalAnswer', "스킬셋 API 호출 과정에서 오류가 발생했습니다.") if final_answer == "스킬셋 API 호출 과정에서 오류가 발생했습니다.": process_view.write("답변 생성에 실패하였습니다.") answer_status.update(label="답변 생성 실패", state="error", expanded=False) else: process_view.write("답변 생성이 완료되었습니다.") answer_status.update(label="답변 생성 완료", state="complete", expanded=False) return final_answer 또는 다음과 같이 상태 코드(200, 400, 500 등)별 오류 처리를 상세하게 구현할 수 있습니다. 사용자에게 각 오류 상황에 대한 메시지 및 에러코드 제공하고, 성공 또는 실패 상태를 UI에 자세히 표시해 줍니다. def get_skillset(query, chat_history=None): # ... (기존 코드) response = requests.post(url, headers=headers, json=data) if response.status_code == 200: return response.json() elif response.status_code == 400: return { 'result': '스킬 클라이언트 오류 발생', 'detail': response.json().get('status').get('code') } elif response.status_code == 500: return { 'result': '스킬 서버 오류 발생', 'detail': response.json().get('status').get('code') } else: return { 'result': '알 수 없는 오류 발생', 'detail': response.status_code } def generate_skillset_response(query, chat_history): with st.status("답변 생성 중...", expanded=True) as answer_status: # ... (기존 코드) result = get_skillset(query, chat_history) status_detail = result.get('detail', None) if result.get('result') == '스킬 클라이언트 오류 발생': final_answer = f"요청이 잘못되었습니다. 입력 내용을 확인하고 다시 시도하세요. (에러 코드: {status_detail})" elif result.get('result') == '스킬 서버 오류 발생': final_answer = f"서버 오류가 발생했습니다. 잠시 후 다시 시도해주세요. (에러 코드: {status_detail})" elif result.get('result') == '알 수 없는 오류 발생': final_answer = f"알 수 없는 문제가 발생했습니다. (상태 코드: {status_detail})" else: final_answer = result.get('result', {}).get('finalAnswer') if result.get('result') in ['스킬 클라이언트 오류 발생', '스킬 서버 오류 발생', '알 수 없는 오류 발생']: process_view.write(f"에러 발생: {final_answer}") answer_status.update(label="답변 생성 실패", state="error", expanded=False) else: process_view.write("답변 생성이 완료되었습니다.") answer_status.update(label="답변 생성 완료", state="complete", expanded=False) return final_answer 관련하여 CLOVA Studio API에서 발생할 수 있는 오류 코드에 대한 원인 및 해결 방안은 CLOVA Studio 문제 해결을 참고해 주세요. 2. 네트워크 오류 재시도 로직을 통해 네트워크 오류나 시간 초과 등의 상황을 대응할 수 있습니다. 요청이 실패할 경우, 지정된 횟수만큼 지연(delay)을 두고 재시도하며, 최종적으로 실패 시 None을 반환하여 후속 처리를 가능하게 합니다. def get_access_token(retries=3, delay=2): # ... (기존 코드) for attempt in range(retries): try: response = requests.get(url, headers=headers) response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: print(f"액세스 토큰 발급 실패 (시도 {attempt + 1}/{retries}): {e}") if attempt < retries - 1: time.sleep(delay) else: return None 마무리 CLOVA Studio의 스킬 트레이너(스킬셋과 라우터), Chat Completions API를 결합하여 지역 검색 에이전트를 구축하는 방법을 알아보았습니다. 이번 쿡북을 통해 에이전트의 동작 원리와 구현 방법을 이해하고, 직접 응용하여 새로운 AI 에이전트를 만들어 보시기 바랍니다!🚀

라우터는 자연어 설명만으로 다양한 입력을 정교하게 다룰 수 있다는 점에서 강점을 가진 솔루션입니다. 이 글에서는 라우터를 실제 서비스에 적용하기 위해 꼭 알아야 할 설계 및 평가의 노하우를 정리했습니다. 라우터의 기본 구조와 활용에 대한 개요는 CLOVA Studio 포럼(LLM 라우터를 활용한 유연한 경로 설계: 분류부터 필터링까지 손쉽게)과 CLOVA Tech Blog(라우터)를 참고하세요. 라우터의 성능을 만드는 작은 차이들 라우터의 정확도는 도메인과 필터를 어떻게 설명하는가에 달려 있습니다. 케이스에 따라 최적의 전략은 다를 수 있으므로, 아래의 설계 팁을 참고해 여러 번 테스트하며 다듬어가는 것을 권장합니다. 1. 역할에 맞게 명료한 설명 작성하기 도메인과 필터는 모두 자연어 설명에 기반해 작동하지만, 설계 전략은 서로 다릅니다. 도메인 설명은 일반화하되 경계를 명확히 하는 것이 핵심입니다. 지나치게 포괄적인 설명은 경계가 무너지고, 반대로 너무 좁은 설명은 실제 다양성을 수용하지 못하게 됩니다. 예를 들어 '건강에 관련된 모든 내용을 포함합니다.'라는 기준은 어떤 입력을 포함하고 배제해야 할지 명확하지 않고, '건강, 의료, 의학, 헬스케어, 메디컬, 웰빙에 대한 내용입니다.'는 지나치게 중복되고 불필요할 수 있습니다. 이상적인 설명은 '질병 진단, 치료법, 약물 복용 등 임상적 의료 행위에 대한 내용입니다.'와 같이 일반화할 수 있는 핵심 기준을 제시하면서도, 포함되는 범위와 포함되지 않는 범위를 암묵적으로 구분할 수 있어야 합니다. 필터 설명은 오히려 제한적이고 명시적으로 작성하는 것이 좋습니다. 포함 기준을 명확히 하고 구체적인 표현이나 입력 패턴 중심으로 설명하는 것이 좋고, 예를 들어 '논란이 될 수 있는 내용입니다.'보다는 '정치적 편향, 종교적 주장, 지역 감정을 유발할 수 있는 표현입니다.'처럼 분명하게 작성하는 것이 바람직합니다. 2. 도메인과 필터, 겹치지 않게 구분하기 분류의 기준이 되는 도메인(또는 필터) 간 경계가 명확하지 않으면, 라우터는 어떤 기준으로 분류해야 할지 혼란스러워질 수 있습니다. 특히 제로샷 방식으로 동작하는 라우터는 설명의 차이를 기반으로 판단하기 때문에, 의미적 중첩을 최대한 피해야 합니다. 예를 들어, '의료'와 '생활건강' 도메인이 모두 '허리 통증'에 대해 다룬다고 할 때, '의료'는 치료 및 진단 중심, '생활건강'은 예방과 습관 중심으로 구분할 수 있습니다. 도메인 또는 필터 간 개념적 충돌이 우려되는 경우, '의학적 목적', '비의학적 목적' 같은 메타 기준을 먼저 정의해 두고, 각 설명에서 이를 기준 삼는 것도 좋은 전략입니다. 3. 직관적인 네이밍 사용하기 도메인 및 필터의 설명뿐만 아니라, 이름 자체도 모델의 분류 판단에 영향을 미치는 요소입니다. 이름이 추상적이거나 이중적인 의미를 가지면, 생성형 모델 기반의 라우터는 분류 기준을 혼동할 수 있습니다. 이름은 해당 설명과 의미적으로 자연스럽게 어울려야 하며, 이름만 보고도 대략적인 역할이나 경계를 유추할 수 있어야 합니다. 설명이 아무리 정교하더라도 '도메인1', '카테고리A'와 같은 임의의 이름은 분류 기준을 흐릴 수 있습니다. 도메인의 경우, 사용자 입력의 의도나 주제를 드러내는 내용 기반의 명확한 키워드나 목적 중심의 명칭(예: 식이요법, 심리 상담 등)이 적합합니다. 필터의 경우, 해석의 여지를 줄이고 분명한 기준에 따라 판단할 수 있도록 구체적이고 제한적인 표현(예: OrientalMedicine, RestrictedBrand 등)을 사용하는 것이 좋습니다. 4. 필요에 따라 예시를 활용하기 설명만으로 기대한 성능이 나오지 않거나 특정 유형의 입력에서 성능이 특히 낮은 경우, 예시를 추가해 보조할 수 있습니다. 예시는 두 가지 방식으로 활용할 수 있습니다. 하나는 실제 입력 예시를 통해 어떤 표현이 해당 도메인에 속하는지를 보여주는 방식이고, 다른 하나는 포함 기준을 나열하여 범위를 명시하는 방식입니다. 예를 들어 '의료' 도메인의 경우 '질병의 원인, 진단, 치료법, 약물 정보 등'과 같이 설명만으로 범위를 분명히 하는 동시에, '대상포진 치료 방법은?', '고혈압 진단 기준 알려주세요'처럼 대표적인 입력 예시도 함께 제공하면 성능을 높이는 데 도움이 됩니다. 단, 예시는 너무 많을 경우 편향을 유도할 수 있으므로 1~2개 수준으로 제한하는 것이 좋습니다. 또한 예시는 설명과 일관된 기준을 따르면서도, 길이 제한이나 유사 표현 확장 등을 고려해 작성하는 것이 좋습니다. 라우터 성능을 평가하는 법 LLM 기반 라우터는 자연어 설명에 따라 동작하기 때문에, 설계자의 설명 방식이나 기준 설정에 따라 성능이 크게 달라질 수 있습니다. 따라서 실제 서비스 수준의 품질을 확보하려면, 단순 직관이 아닌 정량적이고 체계적인 평가를 통해 라우터의 분류 성능을 검증하고 개선해 나가는 과정이 필요합니다. 이를 위해서는 먼저 도메인과 필터 기준을 충분히 다듬고, 다양한 입력을 포함한 테스트셋을 구성한 뒤, 라우터의 응답을 자동으로 테스트해 성능을 측정하고 분석하는 절차를 따르는 것이 좋습니다. 이 과정을 반복하면서, 라우터가 실제 서비스 환경에서도 안정적이고 일관된 분류를 수행할 수 있도록 조정할 수 있습니다. 1. 테스트셋 준비하기 서비스에 적용 가능한 수준의 라우터를 만들기 위해서는 먼저 실제 서비스 맥락을 반영한 테스트셋을 준비해야 합니다. 대표적인 입력만 포함하는 것이 아니라, 다양한 상황을 커버할 수 있도록 입력 유형을 체계적으로 나눠 준비해야 합니다. 예를 들어 다음과 같은 유형을 포함하면 좋습니다. 대표 케이스: 자주 등장하는 일반적인 입력(ex. 고혈압 치료 방법 알려줘) 표현 다양화: 동일 의도의 다양한 표현(ex. 고혈압 약 뭐 먹어야 해?, 혈압 낮추는 약 알려줘) 무관한 입력: 해당 도메인(또는 필터)에 속하지 않는 내용(ex. 오늘 날씨 어때?) 비정형 표현: 실제로 발생할 수 있는 철자 오류 등의 비정형의 입력(ex. 고혀랍약 추천해줘) 이처럼 테스트셋은 모델이 어떤 유형의 입력에 강하고 어떤 유형에 취약한지를 진단할 수 있도록 구성되어야 합니다. 또한 테스트셋은 모든 도메인 및 필터별로 최소 수량 이상 확보하여 균형 있게 구성해야 하며, 각 입력에 대해 어느 도메인이 정답인지 명확하게 정의해두는 것이 중요합니다. 그리고 이 정답 기준은 도메인(또는 필터) 설명과 일관되어야 합니다. 2. 벌크로 테스트하기 벌크 테스트를 본격적으로 실행하기 전에는, 반드시 샘플 데이터로 사전에 개별 테스트를 선행하는 것을 권장합니다. 개별 입력에 대해 수차례 테스트하고 설명을 조정하며 라우터의 응답을 안정화시키는 과정을 먼저 거친 후에 대량 데이터로 테스트를 실행하는 것이 좋습니다. 벌크 테스트를 위한 예시 코드는 다음을 참고하세요. import requests import time import pandas as pd from tqdm import tqdm # 0. 라우터 API 정보 ROUTER_ID = "YOUR_ROUTER_ID" ROUTER_VERSION = "YOUR_ROUTER_VERSION" API_KEY = "YOUR_API_KEY" REQUEST_ID = "YOUR_REQUEST_ID" # 1. 라우터 API 호출 함수 def router_api(query, chat_history=None): url = f"https://clovastudio.stream.ntruss.com/testapp/v1/routers/{ROUTER_ID}/versions/{ROUTER_VERSION}/route" headers = { "Authorization": f"Bearer {API_KEY}", "X-NCP-CLOVASTUDIO-REQUEST-ID": REQUEST_ID, "Content-Type": "application/json" } data = {"query": query} if chat_history: data["chatHistory"] = chat_history # 이용량 초과 시 재시도 while True: response = requests.post(url, headers=headers, json=data) if response.status_code == 429: time.sleep(5) continue return response.json() # 2. 테스트셋 구성 testset = [ {"input": "고혈압 치료 방법 알려줘", "domain": "의료"}, {"input": "혈압 낮추는 약 알려줘", "domain": "의료"}, {"input": "오늘 날씨 어때?", "domain": ""}, # 라우터에 필터를 함께 설정한 경우 다음과 같이 테스트셋 구성 # {"input": "디스크 터진 것 같은데 어떡하죠", "domain": "의료", "content": "[]", "safety": "[]"}, # {"input": "소음인한테 좋은 약 추천해주세요", "domain": "의료", "content": "['OrientalMedicine']", "safety": "[]"}, # {"input": "(특정 국가)인들 때문에 전염된 코로나 증상이랑 치료법은?", "domain": "의료", "content": "[]", "safety": "['unethical']"} # ... ] # 3. 테스트 실행 및 결과 저장 results = [] for i, data in enumerate(tqdm(testset)): try: res = router_api(data["input"]) pred_domain = str(res.get("result", {}).get("domain", {}).get("result")) # pred_content = str(res.get("result", {}).get("blockedContent", {}).get("result", [])) # pred_safety = str(res.get("result", {}).get("safety", {}).get("result", [])) results.append({ "input": data["input"], "domain": data["domain"], # "content": data["content"], # "safety": data["safety"], "pred_domain": pred_domain, # "pred_content": pred_content, # "pred_safety": pred_safety, "is_correct_domain": data["domain"] == pred_domain, # "is_correct_content": data["content"] == pred_content, # "is_correct_safety": data["safety"] == pred_safety }) except Exception as e: print(e) # 4. 결과 확인 df = pd.DataFrame(results) print(df) 출력 결과 input domain pred_domain is_correct_domain 0 고혈압 치료 방법 알려줘 의료 의료 True 1 혈압 낮추는 약 알려줘 의료 의료 True 2 오늘 날씨 어때? True 줄 바꿈 활성화텍 3. 테스트 결과 분석하기 테스트 결과를 수집했다면, 다음 단계는 이를 기반으로 성능을 정량적으로 분석하고 개선 포인트를 도출하는 것입니다. 아래는 분류 성능 분석에서 흔히 활용되는 기준입니다. 정탐 (TP, True Positive): 실제로 해당 도메인(또는 필터)에 속하는 입력을 모델이 올바르게 해당 도메인(또는 필터)으로 예측한 경우 오탐 (FP, False Positive): 실제로는 해당 도메인(또는 필터)에 속하지 않지만, 모델이 잘못 해당 도메인(또는 필터)으로 예측한 경우 미탐 (FN, False Negative): 실제로는 해당 도메인(또는 필터)에 속하지만, 모델이 이를 인식하지 못해 예측 결과가 누락된 경우 정확도 (Accuracy): 전체 테스트 입력 중 정답을 맞춘 비율. (정탐 수 ÷ 전체 입력 수) 앞선 '2. 벌크로 테스트하기'에 이어서 성능 지표를 산출하기 위한 예시 코드는 다음을 참고하세요. # 1. 정탐, 오탐, 미탐, 정확도 계산 tp = ((df["domain"] != "") & (df["domain"] == df["pred_domain"])).sum() fp = ((df["pred_domain"] != "") & (df["domain"] != df["pred_domain"])).sum() fn = ((df["domain"] != "") & (df["pred_domain"] == "")).sum() accuracy = round((df["is_correct_domain"].sum()) / len(df), 3) # 2. 결과 출력 print("라우터 성능 지표 요약") print(f"- 정탐(TP): {tp}") print(f"- 오탐(FP): {fp}") print(f"- 미탐(FN): {fn}") print(f"- 정확도(Accuracy): {accuracy * 100:.1f}%") 출력 결과 라우터 성능 지표 요약 - 정탐(TP): 2 - 오탐(FP): 0 - 미탐(FN): 0 - 정확도(Accuracy): 100.0% 오탐이나 미탐된 케이스는 별도로 수집해 도메인/필터 설명을 개선하는 데 활용할 수 있고, 필요 시 예시 문장을 추가하거나 도메인을 재구성하는 것도 고려할 수 있습니다. 이러한 분석-개선 루프를 꾸준히 반복하면, 실제 서비스 환경에서도 라우터가 안정적이고 일관된 결과를 도출할 수 있습니다. 마무리 라우터의 성능을 높이기 위해서는 설명을 개선하고 반복적으로 테스트하는 과정이 핵심입니다. 라우터 설계 → 테스트셋 구성 → 벌크 테스트 → 결과 분석 → 라우터 수정이라는 일련의 사이클을 통해, 라우터의 품질을 점진적으로 끌어올릴 수 있습니다. 이 과정에서 CLOVA Studio에서 제공하는 샘플 라우터를 참고하는 것도 좋습니다. 본 가이드를 참고하여 똑똑한 맞춤형 라우터를 만들어 보세요. 🚀

라우터(Router)란? LLM 에이전트가 사용자의 요청을 안전하고 신뢰할 수 있게 관리하면 사용자와 운영자 모두에게 든든한 지원이 됩니다. 이를 가능하게 하는 주요 도구가 '라우터(Router)' 입니다. 라우터는 에이전트 제작 과정에서 다양한 도구(Function Calling, 스킬 등)를 효율적으로 연결해 최적의 경로를 제시하고, 사용자의 요청과 콘텐츠를 분석해 적합한 루트를 지정해주는 역할을 합니다. 예를 들어, 사용자가 '최신 금융 상품을 추천해줘.'라고 질문하면, 라우터는 이 요청을 금융 도메인으로 분류하고 관련 태스크가 실행되도록 합니다. 반면, '내 계좌번호 알아?'라는 질문에는 민감한 주제로 감지하여 보안 문제를 사전에 예방합니다. 라우터는 사용자의 질문을 분석해 적절한 태스크와 도메인으로 분류하고, 예기치 않은 상황에서도 적합한 경로를 설정해 사용자 경험을 향상시킵니다. 이제 라우터가 어떻게 활용할 수 있는지 더 알아보겠습니다. 라우터의 구성요소 라우터는 LLM 에이전트가 사용자가 지정한 범위 내에서 정교하고 안전하게 동작하도록, 도메인과 도메인 필터(콘텐츠 필터, 세이프티 필터)로 구성됩니다. 1. 도메인 '도메인(domain)'은 특정 주제나 분야를 의미하며 금융, 여행, 수학 교육 등 각기 다른 주제가 도메인이 될 수 있습니다. 이를테면 금융 도메인에 특화된 LLM은 금융 지식과 용어를 학습하여, 여행과 같은 도메인보다 예금, 대출, 환율 등에 대해 깊이 있는 답변을 제공할 수 있습니다. '도메인' 판별은 사용자의 질문이 허용된 주제에 부합하는지 확인하는 기능입니다. 예를 들어 금융 지원 에이전트에 금융 관련 질문이 들어오면, 금융 특화 스킬이나 Function Calling 등 관련 도구를 호출하여 적절한 응답을 제공합니다. 반면, 여행이나 요리 등 금융과 무관한 주제에 대해서는 도구들(Tools)이 호출되지 않도록 제어함으로써 도구 호출 비용을 절감할 수 있습니다. 2. 도메인 필터 2-1. 콘텐츠 필터 '콘텐츠 필터'는 특정 도메인 내에서 예외적으로 제외된 주제를 감지합니다. 예를 들어, 금융 도메인 내에서도 투자 조언이나 매매와 관련된 대화는 다루지 않도록 필터를 설정할 수 있습니다. 이를 통해 허용된 도메인 범위 안에서라도 특정 주제를 처리하지 않도록 제한할 수 있습니다. 2-2. 세이프티 필터 세이프티 필터는 사용자 발화에서 비윤리적이거나 민감한 내용을 감지하는 기본 제공 기능으로, Unethical 필터와 Contentious 필터가 있습니다. 이 필터들은 기본값으로 설정되어 있으며, on/off 옵션으로 손쉽게 적용할 수 있습니다. Unethical 필터는 욕설, 범죄와 같은 비윤리적인 내용을 포함한 발화를 감지하여 적절히 처리하도록 합니다. Contentious 필터는 정치적, 사회적으로 민감한 논쟁 이슈이나 편향된 주장에 대한 발화를 감지하여 처리합니다. 라우터, 이렇게 활용해보세요! 1. 다른 툴과 함께 활용하기_쿼리 라우팅(Query Routing) 라우터를 활용하여 다른 도구(tools)와 연동해 효율적인 에이전트를 구축할 수 있습니다. 예를 들어, 사용자가 금융 에이전트에게 '최신 예금 상품들을 비교해주세요.'라고 요청한 상황을 가정해보겠습니다. ❶ 사용자의 요청이 들어오면 가장 먼저, ❷ 라우터가 실행되어 쿼리가 다음과 같은 내용을 감지합니다. : (1) 적절한 주제인지(도메인 판별) (2) 도메인 내 허용된 주제인지(콘텐츠 필터), (3) 비윤리적이거나 민감한 내용인지(세이프티 필터) 사용자의 쿼리가 라우터를 무사히 통과하면, ❸ 금융 상품 조회 스킬을 사용하여 API로부터 데이터를 받아, ❹ 비교 및 요약한 답변을 생성하고 ❺ 이를 사용자에게 전달합니다. 반대로, ❶ 사용자의 쿼리가 ❷ 라우터에서 허용되지 않는 내용으로 판별되면, 금융 상품 조회 스킬은 실행되지 않고 ❸거절 메세지로 응답합니다. 이처럼 라우터는 관련 없는 쿼리에 대해 다음 단계의 불필요한 스킬 호출을 방지하여 시스템 자원과 비용을 절약할 수 있습니다. 2. 단독으로 활용하기_콘텐츠 라우팅(Contents Routing) 라우터를 사용하여 특정 규칙에 따라 콘텐츠를 분류할 수 있습니다. 예를 들어, 여러 의견이 오가는 금융 커뮤니티에서 비윤리적인 투자 권유, 불법 정보, 혐오 발언 등을 사전에 감지하고 처리할 수 있습니다. 이 경우, 사용자가 금융 커뮤니티에 글을 작성하면 해당 콘텐츠는 먼저 라우터를 거치게 됩니다. 라우터는 사용자가 작성한 콘텐츠가 정의된 규칙(도메인 및 필터)에 부합하는지 검토하며, 투자 조언(예: 특정 주식 추천, 과도한 수익 보장 등)과 같은 민감한 맥락을 감지하고 분류합니다. 만약 업로드된 게시물이나 SNS 댓글이 라우터에서 부적절한 내용으로 확인되면, 라우터는 해당 콘텐츠의 게시를 차단하거나 관리자가 검토할 수 있도록 표시합니다. 이와 같이 라우터는 모델의 입력을 제어하는 기능으로, 다른 도구들과 연계해 활용하면 서비스의 안정성과 효율성을 더욱 높일 수 있습니다. 라우터를 적용함으로써 LLM 에이전트가 각 상황에서 적절한 답변을 제공하도록 도우며 사용자와 운영자 모두에게 보다 안전하고 신뢰할 수 있는 환경을 제공할 수 있습니다.

@전호영님 안녕하세요. 공유주신 API Spec을 확인해본 결과, servers 항목의 url 필드에 로컬 주소(http://localhost:8080)가 입력되어서 해당 오류가 발생된 것으로 보입니다. 따라서 외부에서도 접근이 가능한 도메인 이름 또는 주소로 변경해 주시면 정상적인 호출이 가능합니다. 이후에도 문제가 지속되거나 추가적인 문의가 있으시다면 언제든 남겨주세요! 감사합니다 🙂

@NewLearn 님, 안녕하세요. 토큰 수 초과로 인해 발생된 에러로 확인되며, 토큰 수 절약에 관한 가이드 링크를 첨부드립니다. (링크 열기) 참고로 1개의 데이터 수집에 제한된 토큰 수는 4,096 입니다. 여기에는 유저 쿼리, 모델의 사고 과정(생각, 액션 등), 스킬 정보(API Spec, Manifest) 등이 모두 합산됩니다. CLOVA Studio의 익스플로러 > 토큰 계산기(HCX)를 통해 토큰 수를 계산해 보실 수 있습니다. 또한 에러 메시지에 오류가 있어 조치할 예정입니다. 이용에 불편을 드려서 죄송합니다. 추가적인 문의 사항이 있으시면 언제든지 남겨주시길 바랍니다. 감사합니다 🙂

@모바일님, 말씀 주신 내용에서 후자와 같이 처리해 주셔야 합니다. 즉, 사용자 쿼리에 대해 chat completion api를 호출할 지 skillset api를 호출할 지 판별하는 도구가 앞단에 필요하게 됩니다. 감사합니다!

@모바일 네 맞습니다. 참고로 관련하여서 연내에 스킬 트레이너 내에 가드레일 기능 추가가 예정되어 있으니 참고해 주셔도 좋을 것 같습니다. 감사합니다 🙂

@모바일 네 확인 감사합니다! 스킬셋 API에서는 query가 인입되면 대부분의 경우에 스킬이 호출되고, 예외적으로 부적절한 쿼리(스킬셋과 연관 없거나 필수 파라미터 누락된 경우 등)가 인입된 케이스에 한해서만 가끔 모델이 자체적으로 스킬 호출을 하지 않고 에러 코드를 반환해 줍니다. 따라서 특정 단어를 말하거나 관련 요청을 한 경우에 한해서만 스킬셋 호출이 이루어지도록 한다면 별도의 판별 모델(의도 분류, 특정 단어 감지 등)을 앞단에 구성해 주셔야 할 것 같습니다. 추가적인 문의가 있으시다면 편히 남겨주십시오. 감사합니다 🙂

@모바일님 안녕하세요. 테스트앱 발급 후 스킬셋 API 호출이 되지 않아서 문의 남겨주신 것으로 이해했습니다. 혹시 스킬셋 API 호출 시에 API 응답이 어떻게 반환되었는지 확인해 주실 수 있을까요? (ex. 에러 코드, 에러 문구, 이외 내용) 그리고 "수수께끼 놀이하자" 쿼리에서만 스킬셋 호출이 되지 않은 것인지, 또는 다른 쿼리에서도 동일하게 재현되는지 확인 부탁드립니다. 확인된 내용을 토대로 적절한 가이드 드릴 수 있도록 하겠습니다. 추가적으로 스킬셋 API 요청 구문 샘플도 공유드립니다. 필수 입력 값으로만 구성된 버전의 샘플입니다. 감사합니다 🙂

@juhn3707님 안녕하세요. 해당 에러는 final answer api 호출 시, 관련 없는 유저쿼리나 필수 파라미터를 누락하여 요청한 경우 발생하며, 전달 주신 케이스는 파라미터 누락으로 인한 오류로 예상됩니다.추가적으로, 필수 파라미터 누락 시에도 정상 동작하도록 API를 구성하실 수 있습니다. 우선 필수 파라미터가 비어있을 시 API에서 오류코드(400 등)를 내려주지 않고, 정상 코드로 반환해야하며(200), 이때 어떠한 파라미터가가 누락되었는 지를 답변 내용에 포함할 수 있습니다. (답변에 포함하지 않은 채 “error”로만 내려주어도 무관합니다.) 이렇게 되면 최종 답변 영역에서 모델이 누락된 파라미터를 요청하는 질문을 생성하고, 원하는 형식으로 튜닝 및 학습도 진행할 수 있습니다. 54020 에러코드에 대한 설명은 7월 중 가이드 문서에 추가될 예정이오니 참고 부탁드립니다. 감사합니다.

안녕하세요 @ak68님, 스킬 트레이너 시나리오 작업 중 토큰 수는 총 4096으로 제한되어 있으며, User query, 모델의 사고 과정(생각 및 액션 등), 스킬 정보(API Spec, Manifest), API 응답(관찰 결과) 등이 모두 합산됩니다. 토큰 수 계산은 CLOVA Studio > 익스플로러 > 토큰 게산기(HCX)에서 이용하실 수 있습니다. 허용 토큰 수를 늘리는 것은 불가능 하지만, 토큰 수를 절약할 수 있는 방안을 안내드립니다. 하기 가이드 링크 접속 후 <API Spec 확장 기능> 참고 바랍니다. https://guide.ncloud-docs.com/docs/clovastudio-skill#api-spec-작성 참고로 위 방안은 API 응답이 너무 길어서 토큰 수 초과되는 경우에 한하여 적용 가능합니다. 추가적으로 궁금한 사항 있으시다면 남겨주십시오. 감사합니다 :)

@beans님 안녕하세요, 남겨주신 질문에 대해 다음과 같이 답변드립니다. 1. 시나리오를 통해 학습 진행 후, API를 이용한 "스킬셋 답변 생성"에서의 x-exclude-cot이 적용 여부 → 스킬 트레이너의 시나리오 화면에서는 x-exclude-cot 적용한 필드는 응답으로 노출되지 않고, 스킬셋 답변 생성 API 호출 시에는 x-exclude-cot 상관 없이 모든 응답이 내려오도록 제공하고 있습니다. x-exclude-cot 목적이 시나리오 내 토큰 수 초과 에러를 방지하고 학습 비용을 절약하는 데에 있기 때문에, 실제 학습 데이터를 제작하는 시나리오 화면에서만 적용되고 있습니다. 2. 응답이 긴 API에서 x-exclude-cot 적용을 하여 시나리오 학습 이후에 API 응답 길이에 대해서 제한 유무 → 스킬 트레이너의 시나리오 화면에서는 API 응답(Step 2-2 관찰)을 포함한 모든 필드(유저쿼리, Step 1, ..., 최종답변)를 합하여 4096 토큰 수로 제한되어 있습니다. 또한 학습 전후에 상관없이 시나리오 작업에서 선택한 버전의 스킬 정보가 x-exclude-cot가 포함된 것이라면, x-exclude-cot가 응답에 적용됩니다. 참고로 스킬셋 답변 생성 API 호출 시에는 별도 길이 제한이 없습니다. 답변이 되셨길 바랍니다! 추가적으로 궁금하신 점 있으시면 남겨주십시오. 감사합니다 🙂