전체 활동내역

안녕하세요 @프란군님, HCX-007 추론 모델에서 추론 기능을 켠 경우, 보다 복잡한 작업에 특화되어 있어 적어주신 것과 같은 단순 지시사항에 대해 부연설명을 하는 등의 경향성이 관찰되고 있습니다. 따라서 목적에 맞게 추론 기능 이용 여부를 결정하실 것을 권장드립니다. thinkingContent/thinking_content 필드는 개발 편의성을 위해 전체 추론 과정을 담아 출력하는 내용으로, 최종 답변(content)과 달리 프롬프팅 등을 통한 내용 및 형식 제어에 한계가 있을 수 있습니다. 또한, 별도의 후처리 없이는 최종 사용자에게 전달하지 않으실 것을 권고드립니다. 감사합니다.

안녕하세요. iOS Naver Map SDK 사용하고 있습니다. 네이버 맵을 움직이거나, 클릭 등 인터랙션을 하면 네이버 로고 색상이 반전되는 현상이 있습니다. 이와 관련한 옵션 설정은 찾지 못했습니다. 맵 인터랙션이 발생하더라도, 기존 컬러 그대로 로고 유지할 수 있는 방법 문의드립니다. 감사합니다.

mobile navermap sdk의 경우 geometry shape 정보를 얻는 API는 없습니다. 별도로 위경도 데이터(WGS84) 리스트를 구축해서 PolygonOverlay를 이용하면 지도 위에 오버레이를 표현할 수는 있을 것 같습니다.

답변 확인이 늦어 죄송합니다. 프롬프트는 "개인정보" 라는 글자가 들어갈 경우 "개인정보는 답변해 줄 수 없습니다." 라고만 글자 그대로 대답해줘. 정도의 프롬프트로 테스트 했을때 HCX-007 기준 추론 모드 ON 시 : 여러 답변 사족과 함께 마크다운 형식으로 이루어진 답변. 추론모드 OFF 시 : "개인정보는 답변해 줄 수 없습니다." 라고 답변 확률 90% 이상 추론모드 여부에 따른 프롬프트 반영 사항이 조금 다른것 같아 문의드린 것입니다. temperature 테스트 기준 0.0 ~ 0.2 top_p= 0.7 ~ 1.0 repetition_penalty=1.0 ~ 1.2 로 테스트 해보았습니다. p.s) 그리고 <thinking>의 추론 내용 자체의 말투는 변경 할 수 없는건가요?

안녕하세요 @sang님, 해당 에러는 정상적인 서비스 API 키를 이용하고 계시지만 이용하시고자 하는 모델로 서비스 앱이 없을 때 발생하는데요. 동일한 계정으로 발급받은 서비스 API 키와 등록된 서비스 앱 이용하고 계시는 게 맞는지 한번 확인 부탁드리며, 맞는데도 불구하고 에러가 발생하는 경우 서비스 문의하기를 통해 계정 정보를 공유해주시면 상세 확인 도와드리겠습니다. 감사합니다.

Projection | 네이버 지도 API v3 아 깃허브 페이지로 구현되어 있군요! WGS84 / UTMK 조합인 점 확인했습니다.

현재 가장 나은 실내 투영도가 있는 지도라 아마 이 지도를 사용하는 것으로 방향이 잡힐 것 같습니다. 전시회장에 별도의 건물용 레이어를 추가하고, 자체적으로 GIS 정보를 추가하는 방식으로 진행할 것 같습니다. 지도 위에 적는 추가 레이어가 존재하므로 네이버 지도의 좌표 정보(WGS84, Web Mercator)나, 지도 데이터에 대한 자세한 지형정보를 알려면 어떻게 해야 할까요? 자체적으로 다루는 레이어에는 점이 아니라 쉐이프를 구현해볼 생각입니다. 공간이 저 건물 한정이라 서비스 내 좌표계와 지형정보가 달라도 큰 차이는 안날 것 같습니다만... 아무래도 좌표계와 투영이 다르단 이유로 모양이 이상하면 보기에 안좋을 수 있다 보니 이런 정보가 확인될지 문의드립니다. (openAI로 뭔가 올린건 WGS84로 예시를 들던데, 그게 맞을까요?) ==== 구체적 예시 대략적으로 이런 느낌으로 마커 혹은 쉐이프가 올라갈 것 같습니다.(지도에 마커나 쉐이프는 네이버 지도의 특정 건물정보나, 속성정보등을 요구하지는 않습니다. 다만, 저 사례의 코스트코는 구글 지도 API 축척으로 해결되는 건물단위 위치이나, 제가 생각한 상황은 0.5M 급 실내 공간 이므로 특정 스케일 부터는 별도의 공간 데이터를 표현할 생각입니다. * 혹시라도 이 방법이 현 네이버 지도 API와는 안되는 예시라면, 알려주시면 감사합니다.

들어가며 Function calling은 LLM이 사용자의 요청을 처리하기 위해 특정 도구를 직접 호출할 수 있도록 해주는 기능입니다. 이를 활용해 목적에 맞는 도구를 정의하고, 실시간 정보 처리나 외부 데이터 연동, 복잡한 계산 등 기존의 LLM만으로는 해결하기 어려운 작업까지 수행할 수 있습니다. 이번 쿡북에서는 Function calling을 복잡한 멀티 턴 시나리오에 적용하여, 궁극적으로 나만의 데이터 자동 분석 솔루션을 구현하는 방법을 살펴보겠습니다. Function calling의 기본 개념과 활용법에 대한 보다 자세한 내용은 아래 이전 글들을 참고하시기 바랍니다. 당신의 AI에게 행동을 맡겨라: 스킬과 Function Calling Function calling 활용법: LLM으로 실시간 데이터 호출과 액션 수행까지 AI 어시스턴트 제작 레시피: Function calling 활용하기 작동 원리 멀티 턴 최근에는 Function calling이 단일 호출로 끝나는 것이 아니라, 사용자와 LLM간 여러 번의 대화를 주고받으며 복합적인 목표를 달성하는 멀티 턴(multi-turn) 방식이 새로운 트렌드로 자리 잡고 있습니다. 기존의 단일 호출 방식은 "오늘 날씨 어때?"처럼 한 번의 질문에 한 번의 Function calling으로 답을 얻는 간단한 시나리오에 적합합니다. 하지만 실제 사용자의 요청은 "오늘 날씨 알려주고, 비가 오면 우산 챙기라고 알려줘"와 같이 여러 요청이 포함된 멀티 쿼리 형태 혹은 모델의 답변을 보고, 추가적인 호출을 요청하는 멀티 턴 형태의 경우가 대다수입니다. 멀티 턴 방식은 이러한 대화의 맥락을 유지하면서 여러 Function calling을 순차적 또는 병렬적으로 수행할 수 있게 해주며, 이를 통해 더욱 복잡하고 실용적인 작업을 처리할 수 있습니다. 기본 작동 방식: 사용자의 요청을 받은 LLM은 사용 가능한 도구 목록을 확인한 뒤, 어떤 도구를 어떤 파라미터와 함께 사용해야 할지를 판단합니다. 그리고 그 결과를 JSON 형태로 반환합니다. 개발자는 이를 실제 도구 실행에 연결하고, 실행 결과를 다시 LLM에 전달합니다. 멀티 턴 작동 방식: 멀티 턴의 경우 이러한 과정이 한 번의 요청으로 끝나지 않고, 이후 이어지는 사용자 요청에서도 이전 맥락을 참고하여 기능을 수행합니다. 이 과정은 대화가 종료될 때까지 반복됩니다. 함수 실행 자동화 기존의 Function calling 방식은 LLM이 사용자 요청을 분석하여 함수를 호출하고 필요한 파라미터를 담은 JSON 객체를 반환하는 것에 그칩니다. 이 경우, 사용자는 반환된 JSON을 파싱하여 직접 코드를 실행하고, 그 결과를 다시 LLM에게 전달하는 번거로운 과정을 거쳐야 합니다. 이 과정은 다음과 같은 흐름으로 진행됩니다. 사용자 요청 LLM이 함수 및 파라미터 호출 사용자가 직접 해당 함수 코드 실행 결과값 획득 결과값을 LLM에게 다시 전달 LLM이 최종 답변 생성 하지만 이번 쿡북에서 다룰 방식은 이러한 과정을 자동화합니다. 미리 함수 코드를 정의하고, 모델이 함수를 실행할 수 있도록 설계합니다. 따라서 LLM이 함수 호출을 결정하면, 시스템이 사용자의 개입 없이 함수를 실행하고 그 결과를 LLM에게 다시 전달하는 과정을 전부 자동으로 처리합니다. 기존 방식과 비교한 멀티 턴 기반 자동화 파이프라인 구조는 다음과 같습니다. 시나리오 이번 쿡북에서 다룰 시나리오는 주문 상품에 대한 고객 CS 데이터 분석 작업입니다. 시나리오에서는 LLM의 Function calling 기능을 활용하여 복잡하고 반복적인 데이터 분석 작업을 자동화하고, 사용자가 자연어로 던지는 질문에 따라 데이터 분석 시각화 등 다양한 작업 수행 과정을 보여줍니다. 시나리오에서 설계할 도구는 총 5개입니다. LLM은 사용자의 요청을 바탕으로 이 도구들을 적절하게 호출하여 데이터 분석 파이프라인을 구축합니다. load_data: 분석할 CSV 파일을 불러오는 도구 filter_data: 특정 조건에 맞는 데이터만 추출하는 도구 check_data: 데이터의 품질을 검사하는 도구 analyze_data: 고객 평점을 기준으로 다양한 통계 정보를 계산하는 도구 save_data: 분석이 완료된 데이터를 저장하는 도구 단순히 도구를 한 번 호출하는 것이 아니라, 사용자와의 멀티 턴 대화 과정에서 여러 도구를 순차적으로 호출하며 복합적인 분석 작업을 자동으로 수행하는 과정을 보여주는 데 초점을 맞추고 있습니다. 구체적인 CS 데이터 분석 시나리오 파이프라인은 다음과 같습니다. 분석 데이터 분석할 데이터는 구매한 제품에 대한 고객 문의 내용이 담긴 CSV 파일입니다. 이 데이터는 다음과 같은 정보를 포함하고 있습니다. id: 고유 식별자 date: 문의 작성 날짜 customer_id: 고객 ID product_name: 상품명 cs_category: CS 유형 (예: as, payment , exchange, delivery, return) text: 고객의 구체적인 문의 내용 rating: 서비스 만족도 평점 (1~5점) 시나리오 구현 1. 환경 설정 API 토큰 발급은 CLOVA Studio API 가이드를 참조하세요. 2. 파일 생성 생성할 파일 구성은 다음과 같습니다. cookbook/ ├── main.py # 멀티 턴 대화 시나리오 실행 스크립트입니다. ├── fc_core.py # 핵심 함수들이 모인 모듈입니다. function_call()로 load_data/filter_data/analyze_data/check_data/save_data 함수들을 제공합니다. ├── config.py # API 설정과 함수 실행 우선순위를 정의합니다. ├── tool_schema.py # AI가 사용할 도구들의 스키마를 정의합니다. 2.1 config.py config.py는 CLOVA Studio API 연결과 함수 실행 우선순위를 정의하는 설정 파일입니다. 이 파일은 전체 시스템의 핵심 설정을 중앙 집중식으로 관리하여 유지보수성을 높입니다. 실제 사용을 위해서는 본인의 CLOVA Studio API 키를 입력해야 합니다. API_KEY = "YOUR_API_KEY" API_URL = "https://clovastudio.stream.ntruss.com/testapp/v3/chat-completions/HCX-DASH-002" def get_headers() -> dict: return { "Content-Type": "application/json", "Authorization": API_KEY, } # 함수 실행 우선 순위 정의 FUNCTION_PRIORITY = { "load_data": 0, "filter_data": 1, "check_data": 2, "analyze_data": 3, "save_data": 4, } 2.2 tool_schema.py tool_schema.py 는 LLM이 사용할 수 있는 도구의 스키마를 정의하는 파일입니다. LLM이 어떤 함수를 호출할 수 있고 각 함수가 어떤 파라미터를 받는지 명시합니다. 정의할 함수 정보는 다음과 같습니다. load_data: CSV 파일을 로드하는 기본 함수입니다. 'filename'은 필수 파라미터이며, 같은 폴더 내에 해당 이름을 가진 파일이 존재해야 합니다. filter_data: 다양한 조건으로 데이터를 필터링하는 함수입니다. 'base' 파라미터로 필터 적용 대상을 선택할 수 있으며, 각 파라미터들은 해당 열에 명시된 값들을 제공합니다. check_data: 데이터 품질을 검사하고 정리하는 함수입니다. 'check_type'과 'action'은 필수 파라미터로, 검사 유형과 수행할 작업을 지정합니다. analyze_data: 특정 제품의 평점 통계를 분석하고 월별 시각화 차트를 생성하는 분석 함수입니다. 'product_name'은 필수 파라미터이며, 'bins'는 시각화 도구의 구간 수를 지정합니다. save_data: 필터링된 데이터를 CSV 파일로 저장하는 함수입니다. 'filename'은 저장할 파일의 이름을 지정하는 필수 파라미터입니다. def get_tools() -> list: return [ { "type": "function", "function": { "description": "CSV 파일을 로드하고 데이터 정보를 제공합니다.", "name": "load_data", "parameters": { "properties": { "filename": { "type": "string", "description": "업로드할 CSV 파일명 (예: 'data.csv', 'sales.csv')", } }, "required": ["filename"], "type": "object", }, }, }, { "type": "function", "function": { "description": "사용자가 특정 조건의 데이터를 요청할 때 호출합니다.", "name": "filter_data", "parameters": { "properties": { "base": { "type": "string", "enum": ["filtered", "all"], "description": "필터 적용 대상. 'filtered'는 이전 필터링 결과에 추가 필터를 적용하고, 'all'은 전체 데이터에서 새로 필터링합니다. 기본값은 'filtered'입니다.", }, "product_name": {"type": "string"}, "cs_category": { "enum": [ "as", "payment", "exchange", "delivery", "return", ], "type": "string", }, "date_range": {"type": "string"}, "rating_threshold": {"type": "integer"}, "customer_id": {"type": "string"}, }, "required": [], "type": "object", }, }, }, { "type": "function", "function": { "description": "특정 product의 rating 기술통계와 월별 시각화 차트를 생성합니다.", "name": "analyze_data", "parameters": { "properties": { "product_name": {"type": "string"}, "bins": { "type": "integer", "description": "rating 히스토그램 구간 수 (기본값: 12)", }, }, "required": ["product_name"], "type": "object", }, }, }, { "type": "function", "function": { "description": "사용자가 데이터 품질 검사나 정리를 요청할 때 호출합니다.", "name": "check_data", "parameters": { "properties": { "check_type": { "description": "검사 유형 (duplicates: 중복, missing: 결측값)", "enum": ["duplicates", "missing"], "type": "string", }, "action": { "description": "수행할 작업 (check: 확인만, remove: 제거)", "enum": ["check", "remove"], "type": "string", }, "column": { "description": "검사할 컬럼명 (없으면 전체)", "type": "string", }, }, "required": ["check_type", "action"], "type": "object", }, }, }, { "type": "function", "function": { "description": "사용자가 특정 파일명으로 데이터를 저장하라고 요청할 때 호출합니다. 필터링된 데이터를 CSV 형식으로 저장합니다.", "name": "save_data", "parameters": { "properties": { "filename": { "description": "저장할 CSV 파일명 (예: 'result.csv')", "type": "string", }, }, "required": ["filename"], "type": "object", }, }, }, ] 2.3. fc_core.py fc_core.py는 멀티 턴, Function calling, 함수 실행 자동화 등 전체 시스템의 핵심 기능을 담당하는 파일입니다. 이 파일은 크게 멀티 턴 대화를 위한 컨텍스트 관리, Function calling 처리, 그리고 실행 함수 구현 세 부분으로 나뉩니다. 2.3.1. 멀티 턴 정보 전달 멀티 턴 대화에서 이전 턴의 정보를 다음 턴에 전달하기 위한 컨텍스트 관리 함수들입니다. build_summary 함수는 현재 대화 상태를 요약하여 문자열로 반환합니다. 데이터 로드 정보, 최근 필터링 결과, 데이터 검사 결과 등 함수 실행 결과를 포함하여 LLM이 이전 턴의 작업 내용을 이해할 수 있도록 도와줍니다. append_context 함수는 앞서 생성된 컨텍스트 요약을 메시지 히스토리에 추가합니다. 이 함수는 다음 턴 시작 전에 호출되어 LLM에게 이전 턴의 작업 결과를 전달합니다. import json import os from typing import Any, Dict, List import pandas as pd import numpy as np import requests import matplotlib matplotlib.use('Agg') import matplotlib.pyplot as plt from config import get_headers, API_URL, FUNCTION_PRIORITY current_df = None # load_data로 불러온 원본 데이터를 의미합니다. filtered_df = None # 다른 함수 실행으로 인해 필터링된 데이터를 의미합니다. conversation_state = { "data_loaded": False, "filename": None, "rows": 0, "columns": 0, "last_filter": None, } def build_summary() -> str: parts: List[str] = [] if conversation_state.get("data_loaded"): parts.append( f"데이터 로드됨(file={conversation_state.get('filename')}, rows={conversation_state.get('rows')}, cols={conversation_state.get('columns')})" ) lf = conversation_state.get("last_filter") if lf: conds = ", ".join(lf.get("conditions", [])) if lf.get("conditions") else "-" parts.append( f"최근 필터(count={lf.get('filtered_count')}/{lf.get('total_count')}, conds={conds})" ) lc = conversation_state.get("last_check") if lc: check_type = lc.get("type", "") action = lc.get("action", "") column = lc.get("column", "") removed = lc.get("removed", 0) remaining = lc.get("remaining", 0) if check_type == "missing": parts.append(f"결측값 제거({column}: {removed}개 제거, {remaining}개 남음)") elif check_type == "duplicates": parts.append(f"중복 제거({column}: {removed}개 제거, {remaining}개 남음)") return "[context] " + " | ".join(parts) if parts else "" def append_context(messages: List[Dict[str, Any]]) -> None: summary = build_summary() if summary: messages.append({"role": "assistant", "content": summary}) 2.3.2. Function calling 처리 LLM이 사용자 쿼리를 바탕으로 정의된 도구를 호출하는 Function calling 과정을 설계합니다. execute_function 함수는 LLM이 요청한 함수를 실행하는 역할을 합니다. 함수 이름과 파라미터를 받아서 해당하는 실제 함수를 호출하고 결과를 반환합니다. function_call 함수는 멀티 턴 대화의 함수 호출 과정을 보여주는 핵심 함수입니다. 첫 번째 호출을 통해 message와 tool_schema를 LLM에게 전송합니다. LLM이 요청 함수들을 우선순위에 따라 실행합니다. 함수 실행 결과를 LLM에게 전달하여 최종 응답을 생성합니다. 새로운 메시지와 이전 컨텍스트를 포함하여 두 번째 턴을 실행합니다. 1-3 과정을 반복하여 최종 응답을 생성합니다. def execute_function(func_name: str, args: Dict[str, Any]) -> Dict[str, Any]: if func_name == "load_data": return load_data_file(args.get("filename")) if func_name == "filter_data": return filter_data_func(args) if func_name == "analyze_data": return analyze_data_func(args) if func_name == "check_data": return check_data_func(args) if func_name == "save_data": return save_data_func(args) return {"error": f"알 수 없는 함수: {func_name}"} def function_call(messages: List[Dict[str, Any]]) -> Dict[str, Any] | None: from tool_schema import get_tools payload = {"messages": messages, "tools": get_tools(), "toolChoice": "auto"} # 요청 페이로드 생성 try: response = requests.post(API_URL, headers=get_headers(), json=payload, timeout=30) # Tool call 요청 전송 if response.status_code != 200: return None # 응답 코드가 200이 아닌 경우 오류 반환 result = response.json() # 응답 결과 파싱 message = ( result.get("result", {}).get("message", {}) if "result" in result else (result.get("choices", [{}])[0].get("message", {})) ) if not message: return result tool_calls_raw = message.get("toolCalls", message.get("tool_calls", [])) # Tool call 목록 추출 if not tool_calls_raw: return {"message": message} tool_calls = sorted( tool_calls_raw, key=lambda c: FUNCTION_PRIORITY.get(c.get("function", {}).get("name", ""), 99), ) # 우선순위에 따라 정렬 tool_messages: List[Dict[str, Any]] = [] for i, call in enumerate(tool_calls): func_name = call["function"]["name"] args_str = call["function"]["arguments"] call_id = call.get("id", f"call_{i}") # 실행 로그: 어떤 함수가 어떤 arguments로 호출되는지 출력 print(f"\n 함수 실행 {i+1}: {func_name}") print(f"Arguments: {args_str}") try: args = json.loads(args_str) if isinstance(args_str, str) else args_str function_result = execute_function(func_name, args) # 함수 실행 print("함수 결과:") print(json.dumps(function_result, ensure_ascii=False, indent=2)) tool_messages.append( { "role": "tool", "content": json.dumps(function_result, ensure_ascii=False), "tool_call_id": call_id, } ) except Exception: tool_messages.append( { "role": "tool", "content": json.dumps({"error": "함수 실행 실패"}, ensure_ascii=False), "tool_call_id": call_id, } ) # 두 번째 호출 messages.extend(tool_messages) # 원본 메시지에 tool 실행 결과 추가 second_payload = {"messages": messages, "tools": get_tools(), "toolChoice": "auto"} # 두 번째 호출 페이로드 생성 second_response = requests.post( API_URL, headers=get_headers(), json=second_payload, timeout=30 ) # 두 번째 호출 요청 전송 if second_response.status_code != 200: return {"message": message, "has_tool_calls": True} second_result = second_response.json() # 두 번째 호출 결과 파싱 second_message = ( second_result.get("result", {}).get("message", {}) if "result" in second_result else (second_result.get("choices", [{}])[0].get("message", {})) ) return {"message": second_message} if second_message else {"message": message, "has_tool_calls": True} except Exception: return None 2.3.3. 실행 함수 구현 실제 데이터 처리를 담당하는 도구 함수들을 구현합니다. 각 함수는 LLM이 요청하는 구체적인 작업을 직접 수행합니다. load_data_file 함수는 CSV 파일을 읽어서 current_df에 저장하고, 데이터의 기본 정보를 반환합니다. 파일 존재 여부를 확인하고, 컬럼명과 데이터 타입 정보를 생성하여 사용자가 데이터 구조를 파악할 수 있도록 도와줍니다. filter_data_func 함수는 다양한 조건으로 데이터를 필터링합니다. 제품명, 카테고리, 평점, 고객 ID, 날짜 범위 등 다양한 조건을 적용할 수 있습니다. 필터링 결과는 filtered_df에 저장되고, 상위 3개 레코드를 미리보기로 제공합니다. check_data_func 함수는 데이터 품질을 검사하고 정리합니다. 중복 데이터와 결측값을 검사하거나 제거할 수 있으며, 특정 컬럼에 대해서만 작업을 수행할 수도 있습니다. analyze_data_func 함수는 특정 제품의 평점 통계를 분석하고 월별 시각화 차트를 생성합니다. 평점에 대한 기술통계(평균, 표준편차, 최소값, 최대값 등)를 계산하고, 월별 건수와 월별 평균 평점을 시각화한 두 개의 차트 파일을 생성합니다. save_data_func 함수는 필터링된 데이터를 CSV 파일로 저장합니다. 파일 확장자가 .csv가 아닌 경우 자동으로 추가하며, 저장된 파일의 정보와 함께 완료 메시지를 반환합니다. # 함수 정의 def load_data_file(filename: str) -> Dict[str, Any]: global current_df try: current_dir = os.getcwd() file_path = os.path.join(current_dir, filename) if not os.path.exists(file_path): return {"error": "파일을 찾을 수 없습니다"} current_df = pd.read_csv(file_path) # 컬럼명과 타입 정보 생성 df_info = f"columns:\n" for i, col in enumerate(current_df.columns): dtype = str(current_df[col].dtype) df_info += f" • {col}: {dtype}" if i < len(current_df.columns) - 1: df_info += "\n" conversation_state.update( { "data_loaded": True, "filename": file_path, "rows": int(len(current_df)), "columns": int(len(current_df.columns)), } ) return { "success": True, "filename": file_path, "rows": int(len(current_df)), "columns": int(len(current_df.columns)), "data_info": df_info, } except Exception: return {"error": "CSV 읽기 실패"} def filter_data_func(args: Dict[str, Any]) -> Dict[str, Any]: global current_df, filtered_df try: if current_df is None: return {"error": "먼저 load_data로 데이터를 로드해주세요."} base_target = args.get("base", "filtered") if base_target == "filtered" and filtered_df is not None and len(filtered_df) > 0: df = filtered_df.copy() else: df = current_df.copy() # 필터 적용 filtered = df.copy() conditions: list[str] = [] if "product_name" in args: filtered = filtered[filtered["product_name"] == args["product_name"]] conditions.append(f"product_name: {args['product_name']}") if "cs_category" in args: filtered = filtered[filtered["cs_category"] == args["cs_category"]] conditions.append(f"cs_category: {args['cs_category']}") if "rating_threshold" in args: filtered = filtered[filtered["rating"] >= args["rating_threshold"]] conditions.append(f"rating >= {args['rating_threshold']}") if "customer_id" in args: filtered = filtered[filtered["customer_id"] == args["customer_id"]] conditions.append(f"customer_id: {args['customer_id']}") if "date_range" in args: date_range = args["date_range"] if "1월" in date_range and "2월" in date_range and "date" in filtered.columns: filtered = filtered[filtered["date"].astype(str).str.contains("2025-01|2025-02")] conditions.append("date_range: 1-2월") filtered_df = filtered # 필터된 데이터 미리보기 (처음 3개만 노출) preview_data = [] if len(filtered_df) > 0: # 미리보기 컬럼 구성 preview_cols = ["id", "date", "product_name", "cs_category", "text", "rating"] available_cols = [col for col in preview_cols if col in filtered_df.columns] preview_data = filtered_df[available_cols].head(3).to_dict('records') conversation_state.update( { "last_filter": { "filtered_count": int(len(filtered_df)), "total_count": int(len(df)), "conditions": conditions, "base": base_target, } } ) return { "success": True, "total_count": int(len(df)), "filtered_count": int(len(filtered_df)), "filter_conditions": conditions, "preview_data": preview_data, "message": f"필터링 완료: 전체 {len(df)}개 중 {len(filtered_df)}개 데이터 추출 (필터링된 데이터가 저장되었습니다)", } except Exception: return {"error": "필터링 실패"} def check_data_func(args: Dict[str, Any]) -> Dict[str, Any]: global current_df, filtered_df try: if current_df is None: return {"error": "먼저 load_data로 데이터를 로드해주세요."} # 항상 필터링된 데이터 기준으로 수행. 없으면 오류 반환 if filtered_df is None or len(filtered_df) == 0: return {"error": "필터링된 데이터가 없습니다. 먼저 filter_data를 수행하세요."} df = filtered_df check_type = args.get("check_type", "duplicates") action = args.get("action", "check") column = args.get("column") if check_type == "duplicates": if column: duplicates_count = df[column].duplicated().sum() else: duplicates_count = df.duplicated().sum() if action == "check": result = { "success": True, "check_type": "duplicates", "action": "check", "column": column, "duplicates_found": int(duplicates_count), "total_records": int(len(df)), } return result else: df_clean = df.drop_duplicates(subset=[column]) if column else df.drop_duplicates() # 필터링된 데이터 갱신 filtered_df = df_clean conversation_state.update( { "last_check": { "type": "duplicates", "action": "remove", "column": column, "removed": int(duplicates_count), "remaining": int(len(df_clean)), } } ) return { "success": True, "check_type": "duplicates", "action": "remove", "column": column, "original_records": int(len(df)), "duplicates_removed": int(duplicates_count), "remaining_records": int(len(df_clean)), } if check_type == "missing": if column: # NaN과 빈 문자열 모두 결측치로 처리 missing_count = df[column].isna().sum() + (df[column] == '').sum() else: # 모든 컬럼에서 NaN 또는 빈 문자열이 있는 행 찾기 missing_mask = df.isna().any(axis=1) | (df == '').any(axis=1) missing_count = missing_mask.sum() if action == "check": return { "success": True, "check_type": "missing", "action": "check", "column": column, "missing_found": int(missing_count), "total_records": int(len(df)), } else: if column: # 특정 컬럼에서 NaN과 빈 문자열 제거 df_clean = df[~(df[column].isna() | (df[column] == ''))] else: # 모든 컬럼에서 NaN 또는 빈 문자열이 있는 행 제거 df_clean = df[~(df.isna().any(axis=1) | (df == '').any(axis=1))] # 필터링된 데이터 갱신 filtered_df = df_clean conversation_state.update( { "last_check": { "type": "missing", "action": "remove", "column": column, "removed": int(missing_count), "remaining": int(len(df_clean)), } } ) return { "success": True, "check_type": "missing", "action": "remove", "column": column, "original_records": int(len(df)), "missing_removed": int(missing_count), "remaining_records": int(len(df_clean)), } return {"error": "지원하지 않는 check_type 입니다."} except Exception: return {"error": "데이터 검사 실패"} def analyze_data_func(args: Dict[str, Any]) -> Dict[str, Any]: global current_df, filtered_df try: if current_df is None: return {"error": "먼저 load_data로 데이터를 로드해주세요."} base_df = filtered_df if (filtered_df is not None and len(filtered_df) > 0) else current_df df = base_df product_name = args.get("product_name") if not product_name: return {"error": "product_name 파라미터가 필요합니다."} if "rating" not in df.columns: return {"error": "rating 컬럼이 없습니다."} df_product = df[df["product_name"] == product_name] if len(df_product) == 0: return {"error": f"해당 product 데이터 없음: {product_name}"} desc = df_product["rating"].describe() stats = { "count": int(desc.get("count", 0)), "mean": float(desc.get("mean", 0)) if not pd.isna(desc.get("mean", None)) else 0.0, "std": float(desc.get("std", 0)) if not pd.isna(desc.get("std", None)) else 0.0, "min": float(desc.get("min", 0)) if not pd.isna(desc.get("min", None)) else 0.0, "25%": float(desc.get("25%", 0)) if not pd.isna(desc.get("25%", None)) else 0.0, "50%": float(desc.get("50%", 0)) if not pd.isna(desc.get("50%", None)) else 0.0, "75%": float(desc.get("75%", 0)) if not pd.isna(desc.get("75%", None)) else 0.0, "max": float(desc.get("max", 0)) if not pd.isna(desc.get("max", None)) else 0.0, } # 월별 시각화 데이터 기반 차트 생성 chart_files = [] if "date" in df_product.columns: try: df_product["date"] = pd.to_datetime(df_product["date"]) df_product["month"] = df_product["date"].dt.strftime("%Y-%m") monthly_counts = df_product["month"].value_counts().sort_index() monthly_ratings = df_product.groupby("month")["rating"].mean() # 차트 1: 월별 건수 plt.figure(figsize=(10, 6)) bars = plt.bar(monthly_counts.index, monthly_counts.values, color='skyblue', alpha=0.7) plt.title(f'{product_name} Monthly AS Cases', fontsize=14, fontweight='bold') plt.xlabel('Month') plt.ylabel('Number of Cases') plt.xticks(rotation=45) plt.grid(True, alpha=0.3) y_max = max(monthly_counts.values) if len(monthly_counts) > 0 else 0 offset = max(0.02 * y_max, 0.5) for bar, count in zip(bars, monthly_counts.values): y = max(bar.get_height() - offset, bar.get_height() * 0.5) plt.text( bar.get_x() + bar.get_width() / 2, y, str(count), ha='center', va='top', fontweight='bold' ) plt.tight_layout(pad=1.2) chart_filename_1 = f"{product_name.replace(' ', '_')}_monthly_cases.png" plt.savefig(chart_filename_1, dpi=300) plt.close() # 차트 2: 월별 평균 평점 plt.figure(figsize=(10, 6)) plt.plot(monthly_ratings.index, monthly_ratings.values, marker='o', linewidth=2, markersize=8, color='red') plt.title(f'{product_name} Monthly Average Rating', fontsize=14, fontweight='bold') plt.xlabel('Month') plt.ylabel('Average Rating') plt.xticks(rotation=45) plt.grid(True, alpha=0.3) plt.ylim(0, 5) for x, y in zip(monthly_ratings.index, monthly_ratings.values): plt.text(x, y + 0.1, f'{y:.2f}', ha='center', va='bottom', fontweight='bold') plt.tight_layout(pad=1.2) chart_filename_2 = f"{product_name.replace(' ', '_')}_monthly_ratings.png" plt.savefig(chart_filename_2, dpi=300) plt.close() chart_files = [chart_filename_1, chart_filename_2] except Exception: chart_files = [] return { "success": True, "product_name": product_name, "stats": stats, "records": int(len(df_product)), "chart_files": chart_files, } except Exception: return {"error": "분석 실패"} def save_data_func(args: Dict[str, Any]) -> Dict[str, Any]: global filtered_df try: if filtered_df is None or len(filtered_df) == 0: return {"error": "저장할 데이터가 없습니다. 먼저 filter_data로 데이터를 준비하세요."} filename = args.get("filename") if not filename: return {"error": "filename 파라미터가 필요합니다."} base, ext = os.path.splitext(filename) if ext.lower() != ".csv": filename = f"{base}.csv" filtered_df.to_csv(filename, index=False, encoding="utf-8") return { "success": True, "filename": filename, "rows": int(len(filtered_df)), "cols": int(len(filtered_df.columns)), "data_source": "filtered", "message": f"데이터 저장 완료: {filename} ({len(filtered_df)}개 레코드, {len(filtered_df.columns)}개 컬럼)", } except Exception: return {"error": "저장 실패"} 2.4. main.py main.py는 멀티 턴 대화 시나리오를 실행하는 메인 파일입니다. 시나리오에 맞는 쿼리와 프롬프트를 입력하여 데이터 분석 작업 수행 예시를 제공합니다. from typing import List, Dict, Any from fc_core import function_call, append_context def multiturn_calling() -> None: # 멀티 턴 대화 시나리오 실행 # 시스템 프롬프트 설정 messages: List[Dict[str, Any]] = [ { "role": "system", "content": "당신은 상품 주문에 대한 고객 CS 데이터를 분석하는 AI 어시스턴트입니다. 함수를 호출해서 CSV 파일을 불러오고, 데이터를 필터링하고, 데이터를 검사하고 분석한 뒤 결과를 바탕으로 사용자에게 도움이 되는 답변을 제공하세요. 일반적인 답변은 하지 말고, 항상 적절한 함수를 선택해서 호출해야 합니다. 사용자가 여러 작업을 요청하면 필요한 모든 함수를 호출하세요.", } ] # 1st Turn: CSV 로드 + Smart watch as 필터링 # - 컨텍스트 요약은 턴2에서만 추가하여 중복/과다 메시지 방지 print("\n🔄 턴 1") # 유저 쿼리 작성 user_query_1 = ( "cs_data.csv 파일을 로드해줘. 그리고 Smart watch의 as 관련 데이터만 필터링해줘" ) print(f" User: {user_query_1}") print("-" * 60) messages.append({"role": "user", "content": user_query_1}) result_1 = function_call(messages) # 1차 호출 + 도구 실행 + 2차 호출로 응답 생성까지 수행 # 어시스턴트 자연어 응답은 메시지 히스토리에 추가하지 않음(Tool 결과만으로 충분) print("✅ 턴 1 완료") print('-'*60) # 2nd Turn: 이전 필터링된 Smart watch 데이터를 분석 + 결측값 제거 + CSV 저장 # filtered_df 기준으로 연속적으로 함수가 수행되는지 확인 print("\n🔄 턴 2") # 2번째 유저 쿼리 작성 user_query_2 = ( "text가 빈 값들은 제거하고 월별로 시각화 분석 한 뒤 데이터를 watch_as.csv 파일로 저장해줘." ) print(f" User: {user_query_2}") print('-'*60) append_context(messages) # 1턴 결과를 요약해 모델에 1회만 전달 messages.append({"role": "user", "content": user_query_2}) result_2 = function_call(messages) # 어시스턴트 자연어 응답은 메시지 히스토리에 추가하지 않음 print("✅ 턴 2 완료") def main() -> None: # 멀티 턴 예시를 실행 multiturn_calling() if __name__ == "__main__": main() 3. 파일 실행 앞서 구현한 파일들을 저장한채로 main.py 파일을 실행한 결과는 다음과 같습니다. 🔄 턴 1 User: cs_data.csv 파일을 로드해줘. 그리고 Smart watch의 as 관련 데이터만 필터링해줘 ------------------------------------------------------------ 함수 실행 1: load_data Arguments: {'filename': 'cs_data.csv'} 함수 결과: { "success": true, "filename": "/Users/user/fc/cs_data.csv", "rows": 500, "columns": 7, "data_info": "columns:\n • id: int64\n • date: object\n • customer_id: object\n • product_name: object\n • cs_category: object\n • text: object\n • rating: int64" } 함수 실행 2: filter_data Arguments: {'product_name': 'Smart watch', 'cs_category': 'as'} 함수 결과: { "success": true, "total_count": 500, "filtered_count": 101, "filter_conditions": [ "product_name: Smart watch", "cs_category: as" ], "preview_data": [ { "id": 5, "date": "2025-01-03", "product_name": "Smart watch", "cs_category": "as", "text": "터치가 안 되는데 수리 가능한가요", "rating": 3 }, { "id": 6, "date": "2025-01-03", "product_name": "Smart watch", "cs_category": "as", "text": NaN, "rating": 3 }, { "id": 10, "date": "2025-01-04", "product_name": "Smart watch", "cs_category": "as", "text": "지문인식이 안 되는데 AS 가능한가요", "rating": 3 } ], "message": "필터링 완료: 전체 500개 중 101개 데이터 추출 (필터링된 데이터가 저장되었습니다)" } ✅ 턴 1 완료 ------------------------------------------------------------ 🔄 턴 2 User: text가 빈 값들은 제거하고 Smart watch에 대해서 월별로 시각화 분석 한 뒤 데이터를 watch_as.csv 파일로 저장해줘. ------------------------------------------------------------ 함수 실행 1: check_data Arguments: {'check_type': 'missing', 'action': 'remove', 'column': 'text'} 함수 결과: { "success": true, "check_type": "missing", "action": "remove", "column": "text", "original_records": 101, "missing_removed": 2, "remaining_records": 99 } 함수 실행 2: analyze_data Arguments: {'product_name': 'Smart watch', 'bins': 12} 함수 결과: { "success": true, "product_name": "Smart watch", "stats": { "count": 99, "mean": 2.4343434343434343, "std": 0.9914081529558191, "min": 1.0, "25%": 1.5, "50%": 3.0, "75%": 3.0, "max": 4.0 }, "records": 99, "chart_files": [ "Smart watch_cases.png", "Smart watch_monthly_ratings.png" ] } 함수 실행 3: save_data Arguments: {'filename': 'watch_as.csv'} 함수 결과: { "success": true, "filename": "watch_as.csv", "rows": 99, "cols": 7, "data_source": "filtered", "message": "데이터 저장 완료: watch_as.csv (99개 레코드, 7개 컬럼)" } ✅ 턴 2 완료 결과 데이터 및 시각화 이미지 watch_as.csv 마무리 이번 쿡북에서는 LLM의 핵심 기능인 Function calling 기반 멀티 턴 자동화를 살펴보았습니다. 시나리오에서는 기본적인 데이터 분석 도구를 사용했지만, 시각화, 감성 분석, 요약 등 다양한 함수를 추가해 더 풍부한 분석을 할 수 있습니다. 예제에서는 원할한 이해를 위해 한 번의 함수 실행으로 멀티 턴을 구현했지만, 실제 환경에서는 단계별 입력을 받아 결과를 확인하고 다음 작업을 결정할 수 있습니다. 이를 통해 필요에 따라 턴을 추가·수정하며 유연한 분석이 가능합니다. 이 쿡북을 통해 Function Calling 역량을 높이고, 자신만의 자동화 솔루션을 구축해 보세요! 🚀

안녕하세요 서비스앱 신청 후 아래처럼 심사완료까지 떳고 서비스앱(테스트api x) api key 도 발급받았는데, 랭체인에서 사용할 때 테스트키는 잘 작동하고, 서비스키만 오류가 발생합니다. 환경변수 CLOVASTUDIO_API_KEY에 서비스앱 키 입력해뒀고, 신청한 모델명으로 아래처럼 호출하고있는데 , 권한 관련 에러가 발생하는 이유가 뭘까요 ? 401 오류가 발생하는 이유와, 정상적으로 호출하기 위해 필요한 추가 설정이 있는지 확인 부탁드립니다. Error code: 401 - {'error': {'message': 'Unauthorized: No Service App. Request for a Service App first.', 'code': '40100'} from langchain_naver import ChatClovaX def get_llm(model='HCX-005'😞 return ChatClovaX(model=model,)

안녕하세요 @hyejeongjo님, 말씀드린대로 현재 CLOVA Studio의 HCX-005 모델로는 이미지 입력과 Function calling을 동시에 이용하실 수 없는데요. 이미지 URL을 포함하는 메시지가 있는 상태에서 tool 메시지를 입력할 경우 이미지에 기반한 Function calling 기능을 활용하는 요청으로 간주되어 에러가 발생할 것으로 보입니다. 해당 제약사항을 감안하되 최대한 이미지에 기반한 Function calling을 구현하는 방법을 찾고 계신 것으로 이해하였습니다. 아래와 같이 두 단계로 나누어 API를 호출하는 방법을 고려해보실 수 있을 것으로 보입니다. 다만 각 단계별로 적절한 시스템 및 유저 프롬프트 작성이 필요할 것으로 보입니다: 1) 이미지 속 정보 추출, 2) 추출된 정보(text) 기반 Function calling image_url 관련 내용을 삭제했을 때 정상 작동하는 케이스를 아래와 같이 알려드립니다. 40009 에러의 경우 보통 이미지 입력 지원을 하지 않는 모델에서 이미지 입력을 했을 때 발생하는 에러인 점 참고 부탁드립니다. # 라이브러리 로드, API 키 설정 등은 완료된 것으로 간주 hcx = ChatClovaX(model="HCX-005", max_tokens=1025) msg = [ SystemMessage(content='당신은 엄격한 JSON 생성기입니다. \n\nCORE RULES:\n- **반드시 JSON 형식으로만 응답하세요. 다른 형식은 절대 사용하지 마세요.**\n- 정확히 하나의 JSON 객체만 출력하세요.\n- 제공된 "spec" 스키마를 정확히 따르세요: 키, 타입, 중첩, 필수 필드, 허용 값.\n- 추가 키, 주석, 설명문, 코드 블록을 사용하지 마세요.\n- 문자열은 유효한 JSON 문자열이어야 합니다: \\" (따옴표), \\\\ (백슬래시), \\\\n (개행), \\\\t (탭).\n- 불린: true/false (소문자). Null: spec에서 허용하는 경우에만 null. 숫자: 유효한 JSON 숫자만.\n- spec에 제시된 키 순서를 존중하세요.\n- 일부 필드가 문서에 누락된 경우, spec과 일치하는 최소한의 유효한 placeholder를 생성하세요:\n * string → ""\n * number → 0\n * boolean → false\n * list → []\n * object → {}\n * Optional → null\n * enum → choose the safest or first allowed value\n- 출력 전에 항상 spec에 대해 출력을 검증하세요. 유효하지 않으면 조용히 재생성하세요.\n- **응답은 반드시 { 로 시작하고 } 로 끝나야 합니다.**\n- **JSON 객체만 출력하고, 다른 것은 출력하지 마세요.**\n- 사용자의 입력을 보고 **도구 목록**을 확인한 후, 아래 지침에 따라 응답하세요.\n\n## 도구 목록\n- get_weather: 날씨 정보를 얻는 도구\n- calculator: 문제를 풀 수 있는 도구. 사용자가 문제 풀이를 원하면 해당 도구를 선택하세요.\n\n### 1. 도구가 필요하다고 판단되는 경우\n- "도구가 필요하다"의 의미: 사용자의 요청이 **도구 목록에 있는 도구로 처리할 수 있는 경우**입니다. \n- 이때는 반드시 JSON 형식으로만 답변하세요. \n- 여러 개 도구를 동시에 선택할 수 있으며, 선택 시 도구 이름(`tool`)만 나열합니다. \n- JSON 외의 텍스트는 절대 포함하지 마세요. \n- 필요한 경우에만 적절한 도구를 선택하세요. \n\n출력 예시:\n{\n "calls": ["get_weather", "search"]\n}\n\n### 2. 도구가 필요하지 않다고 판단되는 경우\n- 도구를 선택하지 않습니다.\n- 아래와 같이 리스트 응답을 비워둡니다.\n\n출력 예시:\n{\n "calls": []\n}\n', additional_kwargs={}, response_metadata={}), HumanMessage(content=[{'type': 'text', 'text': 'x^2-4x+4 = 0 이거 어떻게 풀어'}], additional_kwargs={}, response_metadata={}, id='ee827eb5-344d-4f98-81da-8ae61a09ca4a'), HumanMessage(content='사진을 보고 system prompt의 규칙에 따라 JSON 응답을 생성합니다.', additional_kwargs={}, response_metadata={}, id='972df05c-9911-4ce1-b506-765e6c998efb'), AIMessage(content='', additional_kwargs={'tool_calls': [{'id': 'call_0-1a8a8d6a', 'type': 'function', 'function': {'name': 'calculator', 'arguments': {'problem': '문제 내용 입력', 'explanation': '해설 내용 입력', 'answer': '정답 내용 입력'}}}], 'refusal': None}, response_metadata={'token_usage': {'input_tokens': 1923, 'output_tokens': 10, 'total_tokens': 4335, 'input_token_details': {}, 'output_token_details': {}}}, id='060d1daa-1a6e-43b1-90f9-48223b8b00ba', tool_calls=[{'name': 'calculator', 'args': {'problem': '문제 내용 입력', 'explanation': '해설 내용 입력', 'answer': '정답 내용 입력'}, 'id': 'call_0-1a8a8d6a', 'type': 'tool_call'}]), ToolMessage(content='문제 정답은 정답 내용 입력 이고, 풀이과정은 해설 내용 입력 입니다.', name='calculator', id='2d358e4a-5ad9-4185-96ce-8a61936b0ec7', tool_call_id='call_0-1a8a8d6a'), HumanMessage(content='사진을 보고 system prompt의 규칙에 따라 JSON 응답을 생성합니다.', additional_kwargs={}, response_metadata={}) ] res = hcx.invoke(msg) print(res.content) 위 코드 실행 결과 아래와 같은 출력을 얻을 수 있는 것으로 보입니다: ```json { "result": { "정답": "정답 내용 입력", "풀이과정": "해설 내용 입력" } } ``` 감사합니다.

@CLOVA Studio 운영자9 추가로 아래와 같이 image_url 관련 내용을 전체 삭제했을 경우에도 에러가 발생하는데, > type=image_url로 작성된 HumanMessage 내 content 일부를 삭제하면 정상 작동합니다. 이 부분을 어떻게 수정하셨는지 여쭤보아도 될까요 ? BadRequestError: Error code: 400 - {'error': {'message': 'Unsupported input format', 'code': '40009'}}

@CLOVA Studio 운영자9 네, Function calling 기능이 없어 직접 구현하며 테스트해본 사례인데요. (HCX 내의 function calling이 아닌 여러 call들을 구성하여 이미지 내에서도 가능하도록 테스트 중에 있습니다.) HCX의 function calling 을 사용하지 않아도 메세지 (Image msg+tool message) 를 함께 포함할 수 없는걸까요?

안녕하세요 @hyejeongjo님, 해당 에러는 HCX-005에서 이미지 입력(Vision)과 Function calling 기능을 동시에 이용하고자 하셔서 발생하는 에러로 보입니다. type=image_url로 작성된 HumanMessage 내 content 일부를 삭제하면 정상 작동합니다. 가이드에 명시된 바와 같이 두 기능은 함께 사용하실 수 없는 점 참고 부탁드립니다. 관련하여 에러 코드 및 메시지는 개선할 수 있도록 하겠습니다. 첨언드리자면 말씀주신 것과 같이 이용하실 경우 모델이 사용 가능한 도구 목록에 대한 정의가 되지 않기 때문에 성능이 떨어지는 이슈가 있을 수 있습니다. (CLOVA Studio만의 문제는 아니고, OpenAI 등 타 서비스에서도 마찬가지일 것으로 보입니다.) 따라서 LangChain에서의 일반적인 Tool use 방식으로 사용하실 것을 권장드립니다. 감사합니다.

android 환경에선 새로운 FlutterNaverMap().init() 메서드 사용하면 해결되었습니다

안녕하세요, HCX-005를 활용해 아래 쿼리를 넣었을 때 BadRequestError가 나오는 것을 확인했는데요. 디버깅을 계속 해봐도 어떤 문제인지 알 수 없어서 문의 드립니다. 혹시 입력 값 중 어떤 것이 잘못되었을 까요? (gpt 는 동일 입력 시 정상 작동하여 문의 드립니다.) from langchain_core.messages import AIMessage, AIMessageChunk, HumanMessage, ToolMessage, SystemMessage from langchain_openai import ChatOpenAI from langchain_naver import ChatClovaX hcx = ChatClovaX(model="hcx-005", temperature=0, api_key="nv-xx") msg = [ SystemMessage(content='당신은 엄격한 JSON 생성기입니다. \n\nCORE RULES:\n- **반드시 JSON 형식으로만 응답하세요. 다른 형식은 절대 사용하지 마세요.**\n- 정확히 하나의 JSON 객체만 출력하세요.\n- 제공된 "spec" 스키마를 정확히 따르세요: 키, 타입, 중첩, 필수 필드, 허용 값.\n- 추가 키, 주석, 설명문, 코드 블록을 사용하지 마세요.\n- 문자열은 유효한 JSON 문자열이어야 합니다: \\" (따옴표), \\\\ (백슬래시), \\\\n (개행), \\\\t (탭).\n- 불린: true/false (소문자). Null: spec에서 허용하는 경우에만 null. 숫자: 유효한 JSON 숫자만.\n- spec에 제시된 키 순서를 존중하세요.\n- 일부 필드가 문서에 누락된 경우, spec과 일치하는 최소한의 유효한 placeholder를 생성하세요:\n * string → ""\n * number → 0\n * boolean → false\n * list → []\n * object → {}\n * Optional → null\n * enum → choose the safest or first allowed value\n- 출력 전에 항상 spec에 대해 출력을 검증하세요. 유효하지 않으면 조용히 재생성하세요.\n- **응답은 반드시 { 로 시작하고 } 로 끝나야 합니다.**\n- **JSON 객체만 출력하고, 다른 것은 출력하지 마세요.**\n- 사용자의 입력을 보고 **도구 목록**을 확인한 후, 아래 지침에 따라 응답하세요.\n\n## 도구 목록\n- get_weather: 날씨 정보를 얻는 도구\n- calculator: 문제를 풀 수 있는 도구. 사용자가 문제 풀이를 원하면 해당 도구를 선택하세요.\n\n### 1. 도구가 필요하다고 판단되는 경우\n- "도구가 필요하다"의 의미: 사용자의 요청이 **도구 목록에 있는 도구로 처리할 수 있는 경우**입니다. \n- 이때는 반드시 JSON 형식으로만 답변하세요. \n- 여러 개 도구를 동시에 선택할 수 있으며, 선택 시 도구 이름(`tool`)만 나열합니다. \n- JSON 외의 텍스트는 절대 포함하지 마세요. \n- 필요한 경우에만 적절한 도구를 선택하세요. \n\n출력 예시:\n{\n "calls": ["get_weather", "search"]\n}\n\n### 2. 도구가 필요하지 않다고 판단되는 경우\n- 도구를 선택하지 않습니다.\n- 아래와 같이 리스트 응답을 비워둡니다.\n\n출력 예시:\n{\n "calls": []\n}\n', additional_kwargs={}, response_metadata={}), HumanMessage(content=[{'type': 'text', 'text': 'x^2-4x+4 = 0 이거 어떻게 풀어'}, {'type': 'image_url', 'image_url': {'url': 'https://dimg.donga.com/wps/NEWS/IMAGE/2014/11/17/67941705.3.jpg'}}], additional_kwargs={}, response_metadata={}, id='ee827eb5-344d-4f98-81da-8ae61a09ca4a'), HumanMessage(content='사진을 보고 system prompt의 규칙에 따라 JSON 응답을 생성합니다.', additional_kwargs={}, response_metadata={}, id='972df05c-9911-4ce1-b506-765e6c998efb'), AIMessage(content='', additional_kwargs={'tool_calls': [{'id': 'call_0-1a8a8d6a', 'type': 'function', 'function': {'name': 'calculator', 'arguments': {'problem': '문제 내용 입력', 'explanation': '해설 내용 입력', 'answer': '정답 내용 입력'}}}], 'refusal': None}, response_metadata={'token_usage': {'input_tokens': 1923, 'output_tokens': 10, 'total_tokens': 4335, 'input_token_details': {}, 'output_token_details': {}}}, id='060d1daa-1a6e-43b1-90f9-48223b8b00ba', tool_calls=[{'name': 'calculator', 'args': {'problem': '문제 내용 입력', 'explanation': '해설 내용 입력', 'answer': '정답 내용 입력'}, 'id': 'call_0-1a8a8d6a', 'type': 'tool_call'}]), ToolMessage(content='문제 정답은 정답 내용 입력 이고, 풀이과정은 해설 내용 입력 입니다.', name='calculator', id='2d358e4a-5ad9-4185-96ce-8a61936b0ec7', tool_call_id='call_0-1a8a8d6a'), HumanMessage(content='사진을 보고 system prompt의 규칙에 따라 JSON 응답을 생성합니다.', additional_kwargs={}, response_metadata={}) ] hcx.invoke(msg) 결과: File ~/Desktop/WORKS/edu-max-model/.venv/lib/python3.11/site-packages/openai/_base_client.py:1047, in SyncAPIClient.request(self, cast_to, options, stream, stream_cls) 1044 err.response.read() 1046 log.debug("Re-raising status error") -> 1047 raise self._make_status_error_from_response(err.response) from None 1049 break 1051 assert response is not None, "could not resolve response (should never happen)" BadRequestError: Error code: 400 - {'error': {'message': 'Bad request', 'code': '40000'}}

안녕하세요 @프란군님, HCX-007 모델의 추론 기능을 켜고 이용했을 때 시스템 프롬프트의 지시사항을 따르지 않는 경향성이 있다고 제보주신 것으로 이해했습니다. 맞을까요? 일반적으로 HCX-007 모델은 추론 기능 이용 여부와 무관하게 이전 모델들보다 지시사항 수행 능력이 향상되었으나, 경우에 따라 추론 길이가 길어지는 등의 사유로 인해 지시사항 일부를 누락할 수는 있을 것 같은데요. 혹시 프롬프트를 공유해주실 수 있다면 보다 자세히 살펴보고 말씀드릴 수 있도록 하겠습니다. 포럼을 통해 공유주시기 어렵다면 이용문의를 통해 전달주시면 됩니다. (문의 유형에서 AI Services > CLOVA Studio 선택) 감사합니다.

자답입니만, 추가 문의를 댓글에 남깁니다. 3.19.1에 반영되어있다고 하나, agp 버전에 따른 문제인지 재문의 드립니다.

안녕하세요. 안드로이드 개발자입니다. 금일 구글 플레이 콘솔로부터 내년까지 16kb 페이지 크기 지원 안내를 받았습니다. 아래 구글 가이드대로 확인해보니 https://developer.android.com/guide/practices/page-sizes?hl=ko 아래와 같이 libnavermap.so & libandroid.xgraphics.path.so 가 있어서 안내 받은 것으로 확인되었습니다. 해당 내용에 대한 대응이 별도로 진행되는지 또는 사용자 측에서 추가 처리가 필요한지 가이드를 요청드립니다.

HCX-007을 Langchain을 통해 사용하고 있습니다. 프롬프트 가이드를 작성하고 기존의 HCX-005, HCX-003 과 비교 해봤을때 추론모델은 비교적 프롬프트 주입을 좀 무시하는 경향이 있는것 같습니다. (동일한 프롬프트 기반으로 작성 테스트 기준) 추론 on/off 에 따른 프롬프트 내용의 반응성 특별히 주의해야할 사항이 있는지 궁금해서 문의드립니다. 모델 파라메터값은 아래와 같이 사용하고 있습니다. temperature=0.2, max_completion_tokens=10240, top_p=0.85, repetition_penalty=1.1, 일부 소스: prompt_template = ChatPromptTemplate.from_messages([ SystemMessagePromptTemplate.from_template(system_message_01_combined), MessagesPlaceholder(variable_name="history"), HumanMessagePromptTemplate.from_template(human_template_01_combined) ]) _llm = llm if request.think_mode else nothink_llm chain = prompt_template | _llm # LCEL Chain async for chunk in chain.astream({ "input": request.query, "context": docs_content, "history": messages }, config=RunnableConfig(tags=["final_response_generation"], timeout=120)): #print(f"Chunk data : {chunk}")

예제 페이지에서 확인한 결과 정상 동작합니다. 'init' 이벤트 이후에 호출하신게 맞을까요? - https://navermaps.github.io/maps.js.ncp/docs/tutorial-1-gl-simple.example.html

안녕하세요, /raster-cors 를 이용한 HTTP Referer 인증 방식으로 웹페이지에 static map을 출력하고싶은데요, 공식문서에 나오는 <img src="https://maps.apigw.ntruss.com/map-static/v2/raster-cors?w=300&h=300&center=127.1054221,37.3591614&level=16&X-NCP-APIGW-API-KEY-ID={API Gateway API Key ID}"> 이 코드 그대로, 뒤에 붙은 {API Gateway API Key ID} 만 수정해서 테스트하는데, 출력이 안되고 엑박 이미지깨짐이 발생하네요... 네트워크탭 디버깅을 보면 401 (Unauthorized) 로 출력됩니다. 당연히 네이버 클라우드 플랫폼에서 maps api 사용중이며, 애플리케이션 등록하고 static map으로 등록, Web 서비스 URL도 하나는 실제 웹사이트, 다른 하나는 로컬호스트로 등록 잘 되어있습니다. https://XXX.com http://localhost AWS S3 + cloudfront로 빌드파일 실행중인 웹사이트라, /raster 방식 말고 raster-cors 방식으로 작성하고싶은데, 해당 문제에 대해 해결책을 알고 계신 분 있으신가요? 아님 제가 어떤 치명적인 걸 놓치고 있는걸까요?

안녕하세요. 현재 네이버 지도를 GL 기반(WebGL) 으로 렌더링하여 사용하고 있습니다. mapRef.current = new window.naver.maps.Map("map", { gl: true, zoom: 8, center: new naver.maps.LatLng(36.5, 127.7), }); 이후, 아래와 같이 setMapTypeId를 사용하여 하이브리드 모드로 전환을 시도했으나 동작하지 않았습니다. mapRef.current.setMapTypeId(naver.maps.MapTypeId.HYBRID); 검색해 본 결과, GL 기반 지도는 일반 지도와 별도의 SDK 엔진을 사용하기 때문에 기존 setMapTypeId 방식으로는 위성지도로 변경이 불가능하다는 자료를 확인했습니다. 👉 질문: GL 기반 지도에서 일반/위성 지도 전환 기능을 구현할 수 있는 방법이 있을까요? 감사합니다.

정정

ncpClientId -> ncpKeyId 로 변경하시면 잘 붙습니다;; 저도 이것때문에 하루 날렸네요 ㅠ

들어가며 지난 쿡북 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이 적절한 도구를 자동 선택하고 실행하는 전체 흐름을 구현해 보았습니다. 지정가 매수, 사용자별 계정 관리, 대화 히스토리 저장 등 다양한 방향으로 확장도 가능하니, 나만의 유용한 애플리케이션으로 발전시켜 보세요!