들어가며: 이커머스 AI 고객 서비스의 탄생
저는 3개월 전 국내 이커머스 스타트업에서 AI 고객 서비스 챗봇을 개발했습니다. 초기에는 단순한 키워드 매칭으로 주문 상태를 조회했지만, 고객이 "나의 주문 상태 좀 확인해줘요"라는 한 마디에 다양한 정보가 필요하다는 걸 깨달았습니다. 배송지 주소 확인, 결제 수단 확인, 교환/반품 이력 조회 등 복합적인 함수 호출이 필요했죠.
이 튜토리얼에서는 HolySheep AI를 활용해 **Tool-calling Agent**를 구현하는 방법을 실제 이커머스 시나리오와 함께 설명드리겠습니다. HolySheep AI는
지금 가입하면 무료 크레딧을 제공하며, Claude Sonnet 4.5 ($15/MTok) 및 GPT-4.1 ($8/MTok)을 단일 API 키로 통합 사용할 수 있습니다.
Tool-calling이란 무엇인가?
Tool-calling(함수 호출)은 LLM이 자연어를 이해하여 미리 정의된 함수를 실행하는 메커니즘입니다. 핵심 흐름은 다음과 같습니다:
사용자 입력 → LLM이 함수 식별 → 파라미터 추출 → 함수 실행 → 결과 반환 → LLM이 자연어로 응답
이 구조의 장점은 명확합니다. LLM이 직접 데이터베이스를 조회하지 않고, 개발자가 정의한 함수를 통해 안전하게 데이터를 가져올 수 있습니다. 또한 함수 실행 결과를 다시 LLM에게 전달하여 최종 응답을 생성할 수 있습니다.
1단계: 기본 환경 설정과 함수 정의
먼저 필요한 패키지를 설치하고 HolySheep AI 연결을 설정합니다. 이커머스 AI 고객 서비스 시나리오에서 주문 조회, 상품 검색, FAQ 답변 함수를 구현해보겠습니다.
# requirements.txt
openai>=1.12.0
python-dotenv>=1.0.0
requests>=2.31.0
# config.py
import os
from dotenv import load_dotenv
load_dotenv()
HolySheep AI 설정 - 글로벌 AI API 게이트웨이
HOLYSHEEP_API_KEY = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
모델 설정 (비용 최적화를 위한 선택적 설정)
MODEL_NAME = "claude-sonnet-4-20250514" # $15/MTok - 정확도 우선
MODEL_NAME = "gpt-4.1" # $8/MTok - 비용 효율적
MODEL_NAME = "deepseek-chat-v3" # $0.42/MTok - 대량 처리용
이제 실제 이커머스 시스템에서 사용할 함수들을 정의합니다. 저는 실제 개발 시 주문 조회 함수의 응답 시간 최적화를 위해 Redis 캐싱을 적용했으며, 이 부분이 서비스 응답 속도를 크게 개선했습니다.
# tools.py
import json
from datetime import datetime, timedelta
from typing import Optional, List, Dict, Any
이커머스 Mock 데이터베이스 (실제 환경에서는 DB 연결)
ORDERS_DB = {
"ORD-2024-001": {
"order_id": "ORD-2024-001",
"customer_id": "CUST-1001",
"product": "무선 헤드폰 프로",
"quantity": 1,
"price": 159000,
"status": "배송중",
"tracking_number": "TRK1234567890",
"estimated_delivery": (datetime.now() + timedelta(days=2)).strftime("%Y-%m-%d"),
"shipping_address": "서울시 강남구 테헤란로 123",
"payment_method": "카드결제"
},
"ORD-2024-002": {
"order_id": "ORD-2024-002",
"customer_id": "CUST-1001",
"product": "스마트 워치 울트라",
"quantity": 1,
"price": 349000,
"status": "배송완료",
"tracking_number": "TRK9876543210",
"delivery_date": (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d"),
"shipping_address": "서울시 강남구 테헤란로 123",
"payment_method": "카카오페이"
}
}
PRODUCTS_DB = [
{"id": "PROD-001", "name": "무선 헤드폰 프로", "price": 159000, "stock": 45, "category": "음향기기"},
{"id": "PROD-002", "name": "스마트 워치 울트라", "price": 349000, "stock": 12, "category": "웨어러블"},
{"id": "PROD-003", "name": "노이즈 캔슬링 이어폰", "price": 89000, "stock": 0, "category": "음향기기"},
{"id": "PROD-004", "name": "태블릿 스탠드铝合金", "price": 45000, "stock": 128, "category": "액세서리"}
]
def get_order_status(order_id: str, customer_id: Optional[str] = None) -> Dict[str, Any]:
"""
주문 상태 조회 함수
Args:
order_id: 주문 ID
customer_id: 고객 ID (보안 검증용)
Returns:
주문 상세 정보 또는 오류 메시지
"""
# 실제 환경에서는 DB 인증 검증 수행
if order_id not in ORDERS_DB:
return {
"success": False,
"error": "주문을 찾을 수 없습니다. 주문 ID를 다시 확인해주세요.",
"order_id": order_id
}
order = ORDERS_DB[order_id]
# 고객 ID 검증 (실제 환경에서 구현)
# if customer_id and order["customer_id"] != customer_id:
# return {"success": False, "error": "해당 주문에 접근权限이 없습니다."}
return {
"success": True,
"order": {
"order_id": order["order_id"],
"product": order["product"],
"status": order["status"],
"tracking_number": order.get("tracking_number", "없음"),
"estimated_delivery": order.get("estimated_delivery", order.get("delivery_date", "없음")),
"shipping_address": order["shipping_address"]
}
}
def search_products(query: str, category: Optional[str] = None, max_price: Optional[int] = None) -> Dict[str, Any]:
"""
상품 검색 함수
Args:
query: 검색어
category: 카테고리 필터 (선택)
max_price: 최대 가격 필터 (선택)
Returns:
검색 결과 목록
"""
results = []
for product in PRODUCTS_DB:
# 검색어 매칭 (상품명 또는 카테고리)
if (query.lower() in product["name"].lower() or
query.lower() in product["category"].lower()):
# 필터 적용
if category and product["category"] != category:
continue
if max_price and product["price"] > max_price:
continue
results.append({
"id": product["id"],
"name": product["name"],
"price": product["price"],
"stock_status": "재고있음" if product["stock"] > 0 else "품절"
})
return {
"success": True,
"count": len(results),
"products": results
}
def get_customer_info(customer_id: str) -> Dict[str, Any]:
"""
고객 정보 조회 함수
Args:
customer_id: 고객 ID
Returns:
고객 정보
"""
if customer_id == "CUST-1001":
return {
"success": True,
"customer": {
"id": customer_id,
"name": "홍길동",
"tier": "VIP",
"total_orders": 15,
"total_spent": 2150000
}
}
return {"success": False, "error": "고객 정보를 찾을 수 없습니다."}
2단계: OpenAI 호환 클라이언트로 Tool-calling Agent 구현
HolySheep AI는 OpenAI 호환 API를 제공하므로, OpenAI SDK를 그대로 사용할 수 있습니다. 저는 Claude Sonnet을 사용할 때 함수 호출 정확도가 가장 높았다는 것을 경험했습니다.
# agent.py
import json
import requests
from typing import List, Dict, Any, Optional
from config import HOLYSHEEP_API_KEY, HOLYSHEEP_BASE_URL, MODEL_NAME
class EcommerceToolAgent:
def __init__(self, api_key: str, base_url: str, model: str):
self.api_key = api_key
self.base_url = base_url
self.model = model
self.conversation_history = []
self.tools = self._define_tools()
def _define_tools(self) -> List[Dict[str, Any]]:
"""LLM이 호출할 수 있는 도구 정의 - Claude/Anthropic 형식"""
return [
{
"name": "get_order_status",
"description": "주문 ID를 사용하여 주문 상태, 배송 정보, 추적 번호를 조회합니다.",
"input_schema": {
"type": "object",
"properties": {
"order_id": {
"type": "string",
"description": "조회할 주문 ID (예: ORD-2024-001)"
},
"customer_id": {
"type": "string",
"description": "고객 ID (보안 검증용, 선택사항)"
}
},
"required": ["order_id"]
}
},
{
"name": "search_products",
"description": "상품명, 카테고리, 가격 범위로 상품을 검색합니다.",
"input_schema": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "검색어"
},
"category": {
"type": "string",
"description": "카테고리 필터 (선택): 음향기기, 웨어러블, 액세서리"
},
"max_price": {
"type": "integer",
"description": "최대 가격 (선택)"
}
},
"required": ["query"]
}
},
{
"name": "get_customer_info",
"description": "고객 ID로 고객 등급, 구매 이력 등 정보를 조회합니다.",
"input_schema": {
"type": "object",
"properties": {
"customer_id": {
"type": "string",
"description": "고객 ID"
}
},
"required": ["customer_id"]
}
}
]
def chat(self, user_message: str, customer_id: Optional[str] = None) -> Dict[str, Any]:
"""
사용자 메시지 처리 및 Tool-calling 실행
Args:
user_message: 사용자 입력
customer_id: 현재 고객 ID
Returns:
최종 응답 및 메타데이터
"""
# 대화 기록에 사용자 메시지 추가
self.conversation_history.append({
"role": "user",
"content": user_message
})
# 첫 번째 요청: 도구 호출 여부 결정
response = self._send_request()
# Tool-calling이 필요할 때까지 반복
max_iterations = 5
iteration = 0
while response.get("stop_reason") == "tool_use" and iteration < max_iterations:
iteration += 1
# 도구 호출 요청 처리
tool_calls = response.get("content", [])
tool_results = []
for tool_call in tool_calls:
if tool_call.get("type") == "tool_use":
tool_name = tool_call["name"]
tool_args = tool_call["input"]
# 고객 ID 자동 추가
if customer_id and "customer_id" not in tool_args:
tool_args["customer_id"] = customer_id
# 함수 실행
result = self._execute_tool(tool_name, tool_args)
tool_results.append({
"type": "tool_result",
"tool_use_id": tool_call["id"],
"content": json.dumps(result, ensure_ascii=False)
})
# 도구 결과送信
self.conversation_history.append({
"role": "assistant",
"content": response.get("content", [])
})
self.conversation_history.append({
"role": "user",
"content": tool_results
})
response = self._send_request()
# 최종 응답 추출
final_content = self._extract_text_response(response)
return {
"response": final_content,
"model": self.model,
"iterations": iteration + 1
}
def _send_request(self) -> Dict[str, Any]:
"""HolySheep AI API에 요청送信"""
url = f"{self.base_url}/messages"
headers = {
"Content-Type": "application/json",
"x-api-key": self.api_key,
"anthropic-version": "2023-06-01",
"anthropic-dangerous-direct-browser-access": "true"
}
payload = {
"model": self.model,
"max_tokens": 1024,
"messages": self.conversation_history[-10:], # 최근 10개 메시지만
"tools": self.tools
}
response = requests.post(url, headers=headers, json=payload, timeout=30)
response.raise_for_status()
return response.json()
def _execute_tool(self, tool_name: str, args: Dict[str, Any]) -> Dict[str, Any]:
"""정의된 도구 실행"""
from tools import get_order_status, search_products, get_customer_info
tool_map = {
"get_order_status": get_order_status,
"search_products": search_products,
"get_customer_info": get_customer_info
}
if tool_name in tool_map:
return tool_map[tool_name](**args)
else:
return {"success": False, "error": f"Unknown tool: {tool_name}"}
def _extract_text_response(self, response: Dict[str, Any]) -> str:
"""응답에서 텍스트 추출"""
content = response.get("content", [])
for block in content:
if block.get("type") == "text":
return block.get("text", "")
return str(response)
사용 예시
if __name__ == "__main__":
agent = EcommerceToolAgent(
api_key=HOLYSHEEP_API_KEY,
base_url=HOLYSHEEP_BASE_URL,
model=MODEL_NAME
)
# 테스트 시나리오
test_queries = [
"ORD-2024-001 주문 상태 좀 확인해주세요",
"음향기기 카테고리에서 20만원 이하 상품 검색해주세요",
"CUST-1001 고객의 정보를 알려주세요"
]
for query in test_queries:
print(f"\n[사용자] {query}")
result = agent.chat(query)
print(f"[AI] {result['response']}")
print(f"[정보] 사용된 모델: {result['model']}, iterations: {result['iterations']}")
3단계: 다중 함수 호출과 결과 체이닝
실제 고객 서비스에서는 단일 함수가 아닌 여러 함수를 연쇄적으로 호출해야 합니다. 예를 들어 "최근 주문한 상품 중 배송중인 것의 현재 위치를 알려줘"와 같은 요청은 주문 조회 → 상태 필터링 → 배송 추적의 순서로 처리됩니다.
# advanced_agent.py
import json
import requests
from typing import List, Dict, Any, Optional, Callable
from config import HOLYSHEEP_API_KEY, HOLYSHEEP_BASE_URL, MODEL_NAME
from tools import get_order_status, search_products, get_customer_info
class AdvancedToolAgent:
"""
고급 Tool-calling Agent
- 다중 함수 동시 호출 지원
- 결과 체이닝 및 조건부 실행
- 세션 관리 및 컨텍스트 유지
"""
def __init__(self, api_key: str, base_url: str, model: str):
self.api_key = api_key
self.base_url = base_url
self.model = model
self.session_data = {
"customer_id": None,
"current_orders": [],
"recent_searches": []
}
self.conversation_history = []
self.tools = self._define_tools()
def _define_tools(self) -> List[Dict[str, Any]]:
"""확장된 도구 정의"""
return [
{
"name": "get_order_status",
"description": "주문 ID를 사용하여 주문 상태, 배송 정보, 추적 번호를 조회합니다. customer_id가 제공되면 해당 고객의 주문인지 검증합니다.",
"input_schema": {
"type": "object",
"properties": {
"order_id": {"type": "string", "description": "주문 ID"},
"customer_id": {"type": "string", "description": "고객 ID (세션에서 자동 제공)"}
},
"required": ["order_id"]
}
},
{
"name": "search_products",
"description": "상품명, 카테고리, 가격 범위로 상품을 검색합니다.",
"input_schema": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "검색어"},
"category": {"type": "string", "description": "카테고리 필터"},
"max_price": {"type": "integer", "description": "최대 가격"}
},
"required": ["query"]
}
},
{
"name": "get_customer_info",
"description": "고객 ID로 고객 정보 조회. 등급, 누적 구매 금액, 주문 수 포함.",
"input_schema": {
"type": "object",
"properties": {
"customer_id": {"type": "string", "description": "고객 ID"}
},
"required": ["customer_id"]
}
},
{
"name": "set_customer_session",
"description": "현재 대화의 고객 세션 정보를 설정합니다. 이후 호출에서 자동으로 사용됩니다.",
"input_schema": {
"type": "object",
"properties": {
"customer_id": {"type": "string", "description": "고객 ID"}
},
"required": ["customer_id"]
}
}
]
def process_message(self, user_message: str) -> Dict[str, Any]:
"""메시지 처리 메인 로직"""
self.conversation_history.append({
"role": "user",
"content": user_message
})
# 시스템 프롬프트에 세션 정보注入
system_context = self._build_system_context()
max_iterations = 10
iteration = 0
while iteration < max_iterations:
iteration += 1
response = self._send_request(system_context)
stop_reason = response.get("stop_reason", "")
if stop_reason == "end_turn":
# 최종 응답
final_text = self._extract_text(response)
return {
"success": True,
"response": final_text,
"iterations": iteration
}
elif stop_reason == "tool_use":
# 도구 호출 처리
tool_calls = response.get("content", [])
tool_results = []
for call in tool_calls:
if call.get("type") == "tool_use":
result = self._handle_tool_call(call)
tool_results.append({
"type": "tool_result",
"tool_use_id": call["id"],
"content": json.dumps(result, ensure_ascii=False)
})
self.conversation_history.append({
"role": "assistant",
"content": response.get("content", [])
})
self.conversation_history.append({
"role": "user",
"content": tool_results
})
else:
return {
"success": False,
"response": f"Unexpected stop_reason: {stop_reason}",
"raw_response": response
}
return {
"success": False,
"response": "최대 반복 횟수 초과"
}
def _handle_tool_call(self, call: Dict[str, Any]) -> Dict[str, Any]:
"""개별 도구 호출 처리"""
tool_name = call["name"]
tool_args = call["input"]
# 세션에서 customer_id 자동注入
if self.session_data["customer_id"] and "customer_id" not in tool_args:
tool_args["customer_id"] = self.session_data["customer_id"]
# set_customer_session은 특별 처리
if tool_name == "set_customer_session":
self.session_data["customer_id"] = tool_args.get("customer_id")
return {"success": True, "message": f"고객 세션 설정 완료: {tool_args['customer_id']}"}
# 일반 도구 실행
from tools import get_order_status, search_products, get_customer_info
tool_map = {
"get_order_status": get_order_status,
"search_products": search_products,
"get_customer_info": get_customer_info
}
if tool_name in tool_map:
result = tool_map[tool_name](**tool_args)
# 결과 캐싱
if tool_name == "get_order_status" and result.get("success"):
self.session_data["current_orders"].append(result["order"])
elif tool_name == "search_products" and result.get("success"):
self.session_data["recent_searches"].extend(result.get("products", []))
return result
return {"success": False, "error": f"Unknown tool: {tool_name}"}
def _build_system_context(self) -> Optional[Dict[str, Any]]:
"""시스템 프롬프트에 컨텍스트 추가"""
if not self.session_data["customer_id"]:
return None
return {
"role": "system",
"content": f"""현재 대화는 고객 세션模式下 진행 중입니다.
고객 ID: {self.session_data['customer_id']}
이전 주문 정보: {json.dumps(self.session_data['current_orders'], ensure_ascii=False)}
최근 검색 결과: {json.dumps(self.session_data['recent_searches'][-3:], ensure_ascii=False)}
이 정보를 기반으로 자연스러운 응답을 생성하세요."""
}
def _send_request(self, system_context: Optional[Dict] = None) -> Dict[str, Any]:
"""API 요청送信"""
url = f"{self.base_url}/messages"
headers = {
"Content-Type": "application/json",
"x-api-key": self.api_key,
"anthropic-version": "2023-06-01",
"anthropic-dangerous-direct-browser-access": "true"
}
messages = self.conversation_history[-15:]
if system_context:
messages.insert(0, system_context)
payload = {
"model": self.model,
"max_tokens": 1500,
"messages": messages,
"tools": self.tools
}
response = requests.post(url, headers=headers, json=payload, timeout=30)
response.raise_for_status()
return response.json()
def _extract_text(self, response: Dict[str, Any]) -> str:
"""텍스트 추출"""
for block in response.get("content", []):
if block.get("type") == "text":
return block.get("text", "")
return ""
복합 시나리오 테스트
if __name__ == "__main__":
agent = AdvancedToolAgent(
api_key=HOLYSHEEP_API_KEY,
base_url=HOLYSHEEP_BASE_URL,
model=MODEL_NAME
)
# 시나리오: 고객 세션 설정 후 복합 查询
scenarios = [
"안녕하세요, 고객 세션을 CUST-1001으로 설정해주세요",
"ORD-2024-001 주문 상태 확인해주세요",
"최근 배송중인 주문의 예상 배송일은 언제인가요?",
"같은 카테고리 상품 중에 더 저렴한 게 있나요?"
]
for i, query in enumerate(scenarios):
print(f"\n{'='*60}")
print(f"[{i+1}] 사용자: {query}")
result = agent.process_message(query)
print(f"[AI] {result['response']}")
print(f"[세션] customer_id: {agent.session_data['customer_id']}")
4단계: 결과 파싱과 파이프라인 구축
저는 실제 서비스에서 함수 호출 결과를 파싱하여 외부 시스템에 전달하는 파이프라인을 구축했습니다. 이커머스 플랫폼의 ERP, CRM, 물류 시스템과 연동하는 과정에서 다음과 같은 패턴이 효과적이었습니다.
# pipeline.py
import json
import re
from typing import Dict, Any, List, Optional, Union
from dataclasses import dataclass, asdict
from datetime import datetime
@dataclass
class ParsedToolResult:
"""파싱된 도구 결과"""
tool_name: str
success: bool
data: Dict[str, Any]
raw_text: str
timestamp: str = ""
@dataclass
class CustomerOrderContext:
"""고객 주문 컨텍스트"""
customer_id: str
customer_tier: str = ""
recent_orders: List[Dict] = None
pending_inquiries: List[Dict] = None
def __post_init__(self):
if self.recent_orders is None:
self.recent_orders = []
if self.pending_inquiries is None:
self.pending_inquiries = []
class ToolResultParser:
"""
Tool-calling 결과 파서
- 구조화된 데이터 추출
- 외부 시스템 연동을 위한 포맷 변환
- 응답 템플릿 생성
"""
def __init__(self):
self.parsers = {
"get_order_status": self._parse_order_status,
"search_products": self._parse_product_search,
"get_customer_info": self._parse_customer_info,
"set_customer_session": self._parse_session_set
}
def parse(self, tool_name: str, raw_result: Union[str, Dict]) -> ParsedToolResult:
"""도구 결과 파싱"""
# 문자열인 경우 JSON으로 파싱
if isinstance(raw_result, str):
try:
data = json.loads(raw_result)
except json.JSONDecodeError:
data = {"raw": raw_result}
else:
data = raw_result
# 전용 파서 호출
parser = self.parsers.get(tool_name, self._default_parser)
parsed_data = parser(data)
return ParsedToolResult(
tool_name=tool_name,
success=data.get("success", True),
data=parsed_data,
raw_text=json.dumps(data, ensure_ascii=False),
timestamp=datetime.now().isoformat()
)
def _parse_order_status(self, data: Dict) -> Dict:
"""주문 상태 파싱"""
if not data.get("success"):
return {
"type": "error",
"message": data.get("error", "주문 조회 실패")
}
order = data.get("order", {})
# 상태 기반 표시 형식
status_emoji = {
"주문확인": "📋",
"배송준비중": "📦",
"배송중": "🚚",
"배송완료": "✅",
"취소됨": "❌"
}
return {
"type": "order_status",
"summary": f"{status_emoji.get(order.get('status', ''), '❓')} {order.get('product', '상품')}",
"details": {
"order_id": order.get("order_id"),
"product": order.get("product"),
"status": order.get("status"),
"tracking": order.get("tracking_number"),
"delivery_date": order.get("estimated_delivery"),
"address": self._mask_address(order.get("shipping_address", ""))
},
"formatted_message": self._format_order_message(order)
}
def _parse_product_search(self, data: Dict) -> Dict:
"""상품 검색 결과 파싱"""
if not data.get("success"):
return {
"type": "error",
"message": data.get("error", "상품 검색 실패")
}
products = data.get("products", [])
return {
"type": "product_search",
"count": data.get("count", 0),
"products": [
{
"id": p["id"],
"name": p["name"],
"price": p["price"],
"price_formatted": f"₩{p['price']:,}",
"stock": p["stock_status"]
}
for p in products
],
"formatted_list": self._format_product_list(products)
}
def _parse_customer_info(self, data: Dict) -> Dict:
"""고객 정보 파싱"""
if not data.get("success"):
return {
"type": "error",
"message": data.get("error", "고객 정보 조회 실패")
}
customer = data.get("customer", {})
# 등급 표시
tier_benefits = {
"일반": "기본 혜택",
"실버": "5% 적립 + 무료배송",
"골드": "10% 적립 + 무료배송 + 우선배송",
"VIP": "15% 적립 + 무료배송 + 우선배송 + 전담 상담사"
}
return {
"type": "customer_info",
"details": {
"id": customer.get("id"),
"name": customer.get("name"),
"tier": customer.get("tier"),
"tier_benefit": tier_benefits.get(customer.get("tier", ""), ""),
"total_orders": customer.get("total_orders"),
"total_spent": customer.get("total_spent"),
"total_spent_formatted": f"₩{customer.get('total_spent', 0):,}"
},
"formatted_message": f"""👤 {customer.get('name')} 님
⭐ 등급: {customer.get('tier')}
📊 총 주문: {customer.get('total_orders')}건
💰 누적 구매: {customer.get('total_spent', 0):,}원
✨ 등급 혜택: {tier_benefits.get(customer.get('tier', ''), '')}"""
}
def _parse_session_set(self, data: Dict) -> Dict:
"""세션 설정 결과 파싱"""
return {
"type": "session",
"success": data.get("success", False),
"message": data.get("message", "")
}
def _default_parser(self, data: Dict) -> Dict:
"""기본 파서"""
return {
"type": "unknown",
"raw_data": data
}
@staticmethod
def _mask_address(address: str) -> str:
"""주소 마스킹 (보안)"""
if not address:
return ""
parts = address.split()
if len(parts) >= 3:
return f"{parts[0]} {parts[1]} ***"
return address
@staticmethod
def _format_order_message(order: Dict) -> str:
"""주문 정보 메시지 포맷"""
status = order.get("status", "알 수 없음")
tracking = order.get("tracking_number", "없음")
delivery = order.get("estimated_delivery", "미정")
message = f"""📦 주문번호: {order.get('order_id')}
🛍️ 상품: {order.get('product')}
🚚 상태: {status}
"""
if tracking != "없음":
message += f"🔍 운송장: {tracking}\n"
if delivery != "없음":
message += f"📅 예상 도착: {delivery}\n"
return message
@staticmethod
def _format_product_list(products: List[Dict]) -> str:
"""상품 목록 포맷"""
if not products:
return "검색 결과가 없습니다."
lines = ["🔍 검색 결과:\n"]
for i, p in enumerate(products[:5], 1):
price = f"₩{p['price']:,}"
stock = "✅" if p["stock_status"] == "재고있음" else "❌"
lines.append(f"{i}. {p['name']} - {price} {stock}")
return "\n".join(lines)
class EcommercePipeline:
"""
이커머스 AI 파이프라인
- Tool-calling 결과 처리
- 외부 시스템 연동 (ERP, CRM, WMS)
"""
def __init__(self):
self.parser = ToolResultParser()
self.context = None
def process_with_context(
self,
tool_name: str,
raw_result: Union[str, Dict],
customer_id: str
) -> Dict[str, Any]:
"""컨텍스트 포함 결과 처리"""
# 결과 파싱
parsed = self.parser.parse(tool_name, raw_result)
# 컨텍스트 업데이트
if customer_id:
if self.context is None or self.context.customer_id != customer_id:
self.context = CustomerOrderContext(customer_id=customer_id)
# 타입별 처리
if parsed.tool_name == "get_order_status" and parsed.success:
self.context.recent_orders.append(parsed.data.get("details", {}))
# 외부 시스템 연동용 포맷
external_format = self._to_external_format(parsed)
return {
"parsed_result": asdict(parsed),
"context_summary": asdict(self.context) if self.context else None,
"external_format": external_format,
"display_message": self._build_display_message(parsed)
}
def _to_external_format(self, parsed: ParsedToolResult) -> Dict[str, Any]:
"""외부 시스템 연동용 포맷 변환"""
if parsed.tool_name == "get_order_status":
return {
"action": "UPDATE_ORDER_STATUS",
"order_id": parsed.data.get("details", {}).get("order_id"),
"status": parsed.data.get("details", {}).get("status"),
"timestamp": parsed.timestamp
}
elif parsed.tool_name == "get_customer_info":
return {
"action": "UPDATE_CUSTOMER_CONTEXT",
"customer_id": parsed.data.get("details", {}).get("id"),
"tier": parsed.data.get("details", {}).get("tier"),
"timestamp": parsed.timestamp
}
return {"action": "UNKNOWN"}
@staticmethod
def _build_display_message(parsed: ParsedToolResult) -> str:
"""표시