가상의 Function Calling 함수를 이용한 실습

pip install openai gradio

import os import json import pandas as pd from datetime import datetime from typing import Dict, List, Optional, Union from openai import OpenAI import gradio as gr # 샘플 데이터 (실제 구현에서는 DB나 API에서 가져올 것입니다) products_data = [     {"상품ID": "P001", "상품명": "프리미엄 강아지 간식 세트", "물류창고ID": "W02", "카테고리": "반려용품", "재고수량": 36, "가격": 32417, "입고일자": "2022-01-14"},     {"상품ID": "P002", "상품명": "고양이 캣타워 XL", "물류창고ID": "W02", "카테고리": "반려용품", "재고수량": 91, "가격": 47747, "입고일자": "2023-03-07"},     {"상품ID": "P003", "상품명": "애완동물 자동 급식기", "물류창고ID": "W02", "카테고리": "반려용품", "재고수량": 10, "가격": 22352, "입고일자": "2022-03-04"},     {"상품ID": "P004", "상품명": "반려동물 이동가방", "물류창고ID": "W03", "카테고리": "반려용품", "재고수량": 99, "가격": 91542, "입고일자": "2021-08-01"},     {"상품ID": "P005", "상품명": "펫 전용 샴푸", "물류창고ID": "W03", "카테고리": "반려용품", "재고수량": 49, "가격": 41276, "입고일자": "2020-12-26"} ] customers_data = [     {"고객ID": "C001", "고객명": "박예준", "생년월일": "2001-04-09", "성별": "M", "전화번호": "017-612-4538", "이메일": "sbag@example.com", "가입일자": "2020-08-10", "등급": "VIP", "포인트": 5010},     {"고객ID": "C002", "고객명": "김미숙", "생년월일": "1975-10-05", "성별": "M", "전화번호": "052-368-8400", "이메일": "ohwang@example.org", "가입일자": "2024-01-23", "등급": "VIP", "포인트": 5771},     {"고객ID": "C003", "고객명": "김예준", "생년월일": "2001-02-20", "성별": "M", "전화번호": "061-192-5591", "이메일": "gimyejun@example.org", "가입일자": "2020-07-10", "등급": "일반", "포인트": 4119},     {"고객ID": "C004", "고객명": "곽하은", "생년월일": "1985-12-19", "성별": "M", "전화번호": "053-366-1601", "이메일": "iyeji@example.org", "가입일자": "2021-06-17", "등급": "VIP", "포인트": 4641},     {"고객ID": "C005", "고객명": "김도윤", "생년월일": "1991-02-01", "성별": "M", "전화번호": "017-173-4341", "이메일": "gimsugja@example.net", "가입일자": "2023-09-17", "등급": "일반", "포인트": 4434} ] orders_data = [     {"주문ID": "O0001", "고객ID": "C084", "상품ID": "P011", "수량": 2, "주문일자": "2025-02-25", "결제수단": "카드", "결제상태": "보류", "배송ID": "D0001", "총주문액": 144210, "사용포인트": 640, "할인액": 10848, "배송비": 0, "최종결제액": 132722},     {"주문ID": "O0002", "고객ID": "C025", "상품ID": "P040", "수량": 1, "주문일자": "2025-01-08", "결제수단": "휴대폰결제", "결제상태": "결제완료", "배송ID": "D0002", "총주문액": 87735, "사용포인트": 130, "할인액": 13728, "배송비": 0, "최종결제액": 73877},     {"주문ID": "O0003", "고객ID": "C064", "상품ID": "P003", "수량": 1, "주문일자": "2025-01-26", "결제수단": "휴대폰결제", "결제상태": "결제완료", "배송ID": "D0003", "총주문액": 22352, "사용포인트": 510, "할인액": 1030, "배송비": 3000, "최종결제액": 23812},     {"주문ID": "O0004", "고객ID": "C015", "상품ID": "P020", "수량": 1, "주문일자": "2025-02-04", "결제수단": "휴대폰결제", "결제상태": "취소", "배송ID": "D0004", "총주문액": 77873, "사용포인트": 841, "할인액": 14649, "배송비": 0, "최종결제액": 62383},     {"주문ID": "O0005", "고객ID": "C032", "상품ID": "P007", "수량": 3, "주문일자": "2025-02-11", "결제수단": "계좌이체", "결제상태": "취소", "배송ID": "D0005", "총주문액": 208998, "사용포인트": 366, "할인액": 2840, "배송비": 0, "최종결제액": 205792} ] deliveries_data = [     {"배송ID": "D0001", "배송상태": "배송중", "출고일자": "2025-03-17", "도착예정일": "2025-03-27", "배송사": "우체국택배", "송장번호": "3885733726"},     {"배송ID": "D0002", "배송상태": "배송중", "출고일자": "2025-03-07", "도착예정일": "2025-03-25", "배송사": "CJ대한통운", "송장번호": "3845805571"},     {"배송ID": "D0003", "배송상태": "배송중", "출고일자": "2025-03-20", "도착예정일": "2025-03-27", "배송사": "우체국택배", "송장번호": "9508993384"},     {"배송ID": "D0004", "배송상태": "배송완료", "출고일자": "2025-01-24", "도착예정일": "2025-03-25", "배송사": "우체국택배", "송장번호": "2156991983"},     {"배송ID": "D0005", "배송상태": "배송중", "출고일자": "2025-01-23", "도착예정일": "2025-03-28", "배송사": "한진택배", "송장번호": "8878855674"} ] # 데이터프레임 생성 products_df = pd.DataFrame(products_data) customers_df = pd.DataFrame(customers_data) orders_df = pd.DataFrame(orders_data) deliveries_df = pd.DataFrame(deliveries_data) # 1. 고객 프로필 정보 조회 함수 def get_customer_profile(customer_id: str) -> Dict:     """     고객 ID로 고객 프로필 정보를 조회합니다.     Args:         customer_id: 고객 ID     Returns:         고객 프로필 정보를 담은 사전 (없으면 빈 사전 반환)     """     customer = customers_df[customers_df['고객ID'] == customer_id]     if customer.empty:         return {}     return customer.iloc[0].to_dict() # 2. 고객 주문 내역 조회 함수 def get_customer_orders(customer_id: str, start_date: Optional[str] = None, end_date: Optional[str] = None) -> List[Dict]:     """     고객 ID로 주문 내역을 조회합니다. 선택적으로 날짜 범위를 지정할 수 있습니다.     Args:         customer_id: 고객 ID         start_date: 시작 날짜 (YYYY-MM-DD 형식, 선택사항)         end_date: 종료 날짜 (YYYY-MM-DD 형식, 선택사항)     Returns:         주문 내역 목록 (없으면 빈 리스트 반환)     """     orders = orders_df[orders_df['고객ID'] == customer_id].copy()     if orders.empty:         return []     # 날짜 필터링 (지정된 경우)     if start_date:         orders = orders[orders['주문일자'] >= start_date]     if end_date:         orders = orders[orders['주문일자'] <= end_date]     # 상품 정보 추가     result = []     for _, order in orders.iterrows():         order_dict = order.to_dict()         # 상품 정보 가져오기         product = products_df[products_df['상품ID'] == order['상품ID']]         if not product.empty:             order_dict['상품명'] = product.iloc[0]['상품명']             order_dict['가격'] = product.iloc[0]['가격']         result.append(order_dict)     return result # 3. 고객 배송 상태 조회 함수 def get_customer_deliveries(customer_id: str) -> List[Dict]:     """     고객 ID로 배송 상태를 조회합니다.     Args:         customer_id: 고객 ID     Returns:         배송 정보 목록 (없으면 빈 리스트 반환)     """     # 고객의 주문 ID 가져오기     orders = orders_df[orders_df['고객ID'] == customer_id]     if orders.empty:         return []     # 주문에 해당하는 배송 정보 가져오기     result = []     for _, order in orders.iterrows():         delivery_id = order['배송ID']         delivery = deliveries_df[deliveries_df['배송ID'] == delivery_id]         if not delivery.empty:             delivery_dict = delivery.iloc[0].to_dict()             delivery_dict['주문ID'] = order['주문ID']             delivery_dict['상품ID'] = order['상품ID']             # 상품 정보 추가             product = products_df[products_df['상품ID'] == order['상품ID']]             if not product.empty:                 delivery_dict['상품명'] = product.iloc[0]['상품명']             result.append(delivery_dict)     return result # 4. 고객 포인트 정보 조회 함수 def get_customer_points(customer_id: str) -> Dict:     """     고객 ID로 포인트 정보를 조회합니다.     Args:         customer_id: 고객 ID     Returns:         포인트 정보를 담은 사전 (없으면 빈 사전 반환)     """     customer = customers_df[customers_df['고객ID'] == customer_id]     if customer.empty:         return {}     return {         "고객ID": customer_id,         "고객명": customer.iloc[0]['고객명'],         "등급": customer.iloc[0]['등급'],         "현재포인트": customer.iloc[0]['포인트'],         "가입일자": customer.iloc[0]['가입일자']     } # 5. 결제 정보 조회 함수 def get_payment_info(customer_id: str, order_id: Optional[str] = None) -> Union[Dict, List[Dict]]:     """     고객 ID와 선택적으로 주문 ID로 결제 정보를 조회합니다.     Args:         customer_id: 고객 ID         order_id: 주문 ID (선택사항)     Returns:         결제 정보를 담은 사전 또는 사전 목록 (없으면 빈 사전/리스트 반환)     """     orders = orders_df[orders_df['고객ID'] == customer_id].copy()     if orders.empty:         return {} if order_id else []     # 특정 주문 ID가 지정된 경우     if order_id:         order = orders[orders['주문ID'] == order_id]         if order.empty:             return {}         return {             "주문ID": order_id,             "결제수단": order.iloc[0]['결제수단'],             "결제상태": order.iloc[0]['결제상태'],             "총주문액": order.iloc[0]['총주문액'],             "사용포인트": order.iloc[0]['사용포인트'],             "할인액": order.iloc[0]['할인액'],             "배송비": order.iloc[0]['배송비'],             "최종결제액": order.iloc[0]['최종결제액'],             "주문일자": order.iloc[0]['주문일자']         }     # 모든 주문의 결제 정보 반환     result = []     for _, order in orders.iterrows():         result.append({             "주문ID": order['주문ID'],             "결제수단": order['결제수단'],             "결제상태": order['결제상태'],             "총주문액": order['총주문액'],             "사용포인트": order['사용포인트'],             "할인액": order['할인액'],             "배송비": order['배송비'],             "최종결제액": order['최종결제액'],             "주문일자": order['주문일자']         })     return result # 사용 예시 if __name__ == "__main__":     # 테스트용 고객 ID (실제 구현에서는 로그인한 사용자의 ID를 사용)     customer_id = "C001"     # 프로필 정보 조회     profile = get_customer_profile(customer_id)     print(f"고객 프로필: {profile}")     # 주문 내역 조회     orders = get_customer_orders(customer_id)     print(f"주문 내역: {orders}")     # 배송 상태 조회     deliveries = get_customer_deliveries(customer_id)     print(f"배송 상태: {deliveries}")     # 포인트 정보 조회     points = get_customer_points(customer_id)     print(f"포인트 정보: {points}")     # 결제 정보 조회     payment_info = get_payment_info(customer_id)     print(f"결제 정보: {payment_info}")

고객 프로필: {'고객ID': 'C001', '고객명': '박예준', '생년월일': '2001-04-09', '성별': 'M', '전화번호': '017-612-4538', '이메일': 'sbag@example.com', '가입일자': '2020-08-10', '등급': 'VIP', '포인트': 5010}
주문 내역: []
배송 상태: []
포인트 정보: {'고객ID': 'C001', '고객명': '박예준', '등급': 'VIP', '현재포인트': np.int64(5010), '가입일자': '2020-08-10'}
결제 정보: []

 

# OpenAI 클라이언트 설정 client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY", "여러분의 Key 값"))

# 함수 매핑 - 실제 함수를 호출하는 역할 def execute_function(function_name, arguments):     function_map = {         "get_customer_profile": get_customer_profile,         "get_customer_orders": get_customer_orders,         "get_customer_deliveries": get_customer_deliveries,         "get_customer_points": get_customer_points,         "get_payment_info": get_payment_info     }     if function_name not in function_map:         return {"error": f"Function {function_name} not found"}     try:         # customer_id가 비어있는지 확인         if "customer_id" in arguments and (arguments["customer_id"] is None or arguments["customer_id"] == ""):             arguments["customer_id"] = "C001"  # 기본값 설정         # 함수 실행         result = function_map[function_name](**arguments)         # int64, float64 등의 numpy 타입을 일반 Python 타입으로 변환         if isinstance(result, dict):             for key, value in result.items():                 if hasattr(value, 'item'):  # numpy 숫자 타입 확인                     result[key] = value.item()  # Python 내장 타입으로 변환         elif isinstance(result, list):             for item in result:                 if isinstance(item, dict):                     for key, value in item.items():                         if hasattr(value, 'item'):                             item[key] = value.item()         return result     except Exception as e:         return {"error": str(e)}

# OpenAI 함수 호출 정의 tools = [     {         "type": "function",         "function": {             "name": "get_customer_profile",             "description": "고객 ID로 고객 프로필 정보를 조회합니다.",             "parameters": {                 "type": "object",                 "properties": {                     "customer_id": {                         "type": "string",                         "description": "고객 ID (예: C001)"                     }                 },                 "required": ["customer_id"]             }         }     },     {         "type": "function",         "function": {             "name": "get_customer_orders",             "description": "고객 ID로 주문 내역을 조회합니다. 선택적으로 날짜 범위를 지정할 수 있습니다.",             "parameters": {                 "type": "object",                 "properties": {                     "customer_id": {                         "type": "string",                         "description": "고객 ID (예: C001)"                     },                     "start_date": {                         "type": "string",                         "description": "시작 날짜 (YYYY-MM-DD 형식, 선택사항)"                     },                     "end_date": {                         "type": "string",                         "description": "종료 날짜 (YYYY-MM-DD 형식, 선택사항)"                     }                 },                 "required": ["customer_id"]             }         }     },     {         "type": "function",         "function": {             "name": "get_customer_deliveries",             "description": "고객 ID로 배송 상태를 조회합니다.",             "parameters": {                 "type": "object",                 "properties": {                     "customer_id": {                         "type": "string",                         "description": "고객 ID (예: C001)"                     }                 },                 "required": ["customer_id"]             }         }     },     {         "type": "function",         "function": {             "name": "get_customer_points",             "description": "고객 ID로 포인트 정보를 조회합니다.",             "parameters": {                 "type": "object",                 "properties": {                     "customer_id": {                         "type": "string",                         "description": "고객 ID (예: C001)"                     }                 },                 "required": ["customer_id"]             }         }     },     {         "type": "function",         "function": {             "name": "get_payment_info",             "description": "고객 ID와 선택적으로 주문 ID로 결제 정보를 조회합니다.",             "parameters": {                 "type": "object",                 "properties": {                     "customer_id": {                         "type": "string",                         "description": "고객 ID (예: C001)"                     },                     "order_id": {                         "type": "string",                         "description": "주문 ID (예: O0001, 선택사항)"                     }                 },                 "required": ["customer_id"]             }         }     } ]

# 대화 기록을 저장할 변수 conversation_history = {}

def process_message(message, customer_id, conversation_id=None):     global conversation_history     # 새로운 대화인 경우 대화 ID 생성 및 초기화     if conversation_id is None or conversation_id not in conversation_history:         conversation_id = f"conv_{len(conversation_history) + 1}"         conversation_history[conversation_id] = [             # 시스템 메시지로 고객 ID 컨텍스트 설정             {"role": "system", "content": f"현재 고객 ID는 {customer_id}입니다. 이 고객의 정보를 조회할 때는 항상 이 ID를 사용하세요."}         ]     # 사용자 메시지 추가     conversation_history[conversation_id].append({"role": "user", "content": message})     # 메시지 목록 구성     messages = conversation_history[conversation_id]     try:         # OpenAI API 호출         response = client.chat.completions.create(             model="gpt-4o",  # 또는 사용 가능한 최신 모델             messages=messages,             tools=tools,             tool_choice="auto"         )         # 응답 추출         assistant_message = response.choices[0].message         # 함수 호출이 필요한 경우         if assistant_message.tool_calls:             # 응답에 함수 호출 정보 저장             conversation_history[conversation_id].append(assistant_message)             # 모든 함수 호출 수행             for tool_call in assistant_message.tool_calls:                 function_name = tool_call.function.name                 function_args = json.loads(tool_call.function.arguments)                 # customer_id가 명시되지 않은 경우 현재 고객 ID 사용                 if "customer_id" in function_args and not function_args["customer_id"]:                     function_args["customer_id"] = customer_id                 # 함수 실행 결과 가져오기                 function_response = execute_function(function_name, function_args)                 # 함수 결과를 대화 기록에 추가                 conversation_history[conversation_id].append({                     "role": "tool",                     "tool_call_id": tool_call.id,                     "name": function_name,                     "content": json.dumps(function_response, ensure_ascii=False)                 })             # 함수 결과를 바탕으로 최종 응답 생성             second_response = client.chat.completions.create(                 model="gpt-4o",                 messages=conversation_history[conversation_id]             )             final_response = second_response.choices[0].message.content             # 최종 응답을 대화 기록에 추가             conversation_history[conversation_id].append({                 "role": "assistant",                 "content": final_response             })             return final_response, conversation_id         else:             # 일반 응답인 경우             content = assistant_message.content             # 응답을 대화 기록에 추가             conversation_history[conversation_id].append({                 "role": "assistant",                 "content": content             })             return content, conversation_id     except Exception as e:         error_message = f"오류가 발생했습니다: {str(e)}"         return error_message, conversation_id

def create_ui():     with gr.Blocks(title="고객 서비스 챗봇") as demo:         gr.Markdown("# 고객 서비스 챗봇")         gr.Markdown("고객 ID를 입력하고 문의사항을 입력하세요.")         with gr.Row():             customer_id_input = gr.Textbox(label="고객 ID", placeholder="예: C001", value="C001")         conversation_id = gr.State(value=None)         chatbot = gr.Chatbot(label="대화", height=400)         msg = gr.Textbox(label="메시지", placeholder="무엇을 도와드릴까요?")         def user_input(message, history, customer_id, conv_id):             if not message.strip():                 return gr.update(value=""), history, conv_id             # 고객 ID가 비어있는 경우 기본값 설정             if not customer_id or customer_id.strip() == "":                 customer_id = "C001"  # 기본값 설정             response, new_conv_id = process_message(message, customer_id, conv_id)             history.append((message, response))             return gr.update(value=""), history, new_conv_id         msg.submit(user_input, [msg, chatbot, customer_id_input, conversation_id], [msg, chatbot, conversation_id])         clear = gr.Button("대화 초기화")         def clear_conversation():             return [], None         clear.click(clear_conversation, [], [chatbot, conversation_id])     return demo

# 주피터 노트북 환경에서 UI 실행 demo = create_ui() demo.launch(debug=True)

<ipython-input-9-46953d90f13d>:11: UserWarning: You have not specified a value for the `type` parameter. Defaulting to the 'tuples' format for chatbot messages, but this is deprecated and will be removed in a future version of Gradio. Please set type='messages' instead, which uses openai-style dictionaries with 'role' and 'content' keys.
  chatbot = gr.Chatbot(label="대화", height=400)

Running Gradio in a Colab notebook requires sharing enabled. Automatically setting `share=True` (you can turn this off by setting `share=False` in `launch()` explicitly).

Colab notebook detected. This cell will run indefinitely so that you can see errors and logs. To turn off, set debug=False in launch().
* Running on public URL: https://b2cbf0cd3f94a348e1.gradio.live

This share link expires in 72 hours. For free permanent hosting and GPU upgrades, run `gradio deploy` from the terminal in the working directory to deploy to Hugging Face Spaces (https://huggingface.co/spaces)