에이전트 도구(Tools)와 UC 함수

-- 주문 상태 조회 도구
CREATE OR REPLACE FUNCTION catalog.schema.get_order_status(
    order_id BIGINT COMMENT '조회할 주문번호'
)
RETURNS TABLE(order_id BIGINT, status STRING, shipping_date DATE, estimated_arrival DATE)
COMMENT '주문번호로 현재 주문 상태와 배송 정보를 조회합니다. 주문 상태가 궁금하거나 배송 정보를 확인할 때 사용하세요.'
RETURN
    SELECT order_id, status, shipping_date, estimated_arrival
    FROM catalog.schema.orders
    WHERE order_id = get_order_status.order_id;

-- 고객별 최근 주문 요약 도구
CREATE OR REPLACE FUNCTION catalog.schema.get_customer_summary(
    customer_email STRING COMMENT '고객 이메일 주소'
)
RETURNS TABLE(name STRING, tier STRING, total_orders INT, lifetime_value DECIMAL(12,2), last_order_date DATE)
COMMENT '이메일로 고객 정보와 주문 요약을 조회합니다. 고객 문의 시 기본 정보를 확인하는 데 사용하세요.'
RETURN
    SELECT
        c.name,
        c.tier,
        COUNT(o.order_id) AS total_orders,
        SUM(o.amount) AS lifetime_value,
        MAX(o.order_date) AS last_order_date
    FROM catalog.schema.customers c
    LEFT JOIN catalog.schema.orders o ON c.customer_id = o.customer_id
    WHERE c.email = get_customer_summary.customer_email
    GROUP BY c.name, c.tier;

-- 외부 날씨 API 호출 도구
CREATE OR REPLACE FUNCTION catalog.schema.get_weather(
    city STRING COMMENT '날씨를 조회할 도시 이름 (예: Seoul, Busan)'
)
RETURNS STRING
LANGUAGE PYTHON
COMMENT '도시의 현재 날씨 정보를 조회합니다. 배송 예상이나 야외 이벤트 관련 질문에 사용하세요.'
AS $$
    import requests

    response = requests.get(
        f"https://api.weatherapi.com/v1/current.json",
        params={"key": "API_KEY", "q": city}
    )
    if response.ok:
        data = response.json()
        current = data["current"]
        return f"{city}: {current['temp_c']}°C, {current['condition']['text']}"
    else:
        return f"날씨 정보를 가져올 수 없습니다: {response.status_code}"
$$;

-- 이메일 발송 도구
CREATE OR REPLACE FUNCTION catalog.schema.send_notification(
    recipient_email STRING COMMENT '수신자 이메일 주소',
    subject STRING COMMENT '이메일 제목',
    body STRING COMMENT '이메일 본문'
)
RETURNS STRING
LANGUAGE PYTHON
COMMENT '고객에게 이메일 알림을 발송합니다. 주문 확인, 배송 알림 등에 사용하세요.'
AS $$
    import requests

    response = requests.post(
        "https://api.company.com/notify",
        json={"to": recipient_email, "subject": subject, "body": body},
        headers={"Authorization": "Bearer TOKEN"}
    )
    return f"발송 {'성공' if response.ok else '실패'}: {recipient_email}"
$$;

from databricks.agents.tools import VectorSearchRetrieverTool

# Vector Search 기반 Retriever 도구
retriever_tool = VectorSearchRetrieverTool(
    index_name="catalog.schema.docs_index",
    num_results=5,
    columns=["content", "title", "source_url"],
    query_type="ANN",  # Approximate Nearest Neighbor
    tool_description="회사 내부 문서, FAQ, 제품 매뉴얼에서 관련 정보를 검색합니다. 고객 질문에 답변하기 위한 정보가 필요할 때 사용하세요."
)

from databricks.agents.tools import UCFunctionToolkit

# UC 함수들을 도구 목록으로 변환
toolkit = UCFunctionToolkit(
    function_names=[
        "catalog.schema.get_order_status",
        "catalog.schema.get_customer_summary",
        "catalog.schema.send_notification",
    ]
)

# 도구 목록 확인
for tool in toolkit.tools:
    print(f"Tool: {tool.name}")
    print(f"  Description: {tool.description}")
    print(f"  Parameters: {tool.parameters}")
    print()

from databricks.agents.tools import UCFunctionToolkit
from langchain_community.chat_models import ChatDatabricks
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate

# 1. UC 함수를 LangChain 도구로 변환
toolkit = UCFunctionToolkit(
    function_names=[
        "catalog.schema.get_order_status",
        "catalog.schema.get_customer_summary",
    ]
)
tools = toolkit.tools

# 2. LLM 설정
llm = ChatDatabricks(
    endpoint="databricks-meta-llama-3-3-70b-instruct",
    temperature=0.1
)

# 3. 프롬프트 설정
prompt = ChatPromptTemplate.from_messages([
    ("system", "당신은 고객 지원 에이전트입니다. 제공된 도구를 사용하여 정확하게 답변하세요."),
    ("placeholder", "{chat_history}"),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}")
])

# 4. 에이전트 생성 및 실행
agent = create_tool_calling_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# 5. 실행
response = executor.invoke({"input": "주문 12345의 상태를 알려주세요", "chat_history": []})
print(response["output"])

-- ❌ 나쁜 예: LLM이 언제 이 도구를 써야 하는지 판단하기 어렵습니다
CREATE FUNCTION get_info(id BIGINT)
RETURNS TABLE(...)
COMMENT '정보를 조회합니다'
...

-- ✅ 좋은 예: LLM이 정확히 언제, 어떤 입력으로 호출해야 하는지 알 수 있습니다
CREATE FUNCTION get_order_status(order_id BIGINT COMMENT '주문번호 (숫자, 예: 12345)')
RETURNS TABLE(order_id BIGINT, status STRING, shipping_date DATE)
COMMENT '주문번호로 현재 배송 상태를 조회합니다.
고객이 "내 주문 어디 있어?", "배송 언제 오나요?", "주문 상태 확인" 등의
질문을 할 때 이 도구를 사용하세요.
반품이나 환불 관련 질문에는 이 도구가 아닌 get_return_status를 사용하세요.
order_id를 모르는 경우, 먼저 get_customer_orders 도구로 주문 목록을 조회하세요.'
...

-- 1. 에이전트 전용 서비스 프린시펄 생성
-- (Databricks Account Console에서 수행)

-- 2. 필요한 함수에만 EXECUTE 권한 부여
GRANT EXECUTE ON FUNCTION catalog.schema.get_order_status
  TO `agent-service-principal`;
GRANT EXECUTE ON FUNCTION catalog.schema.get_customer_summary
  TO `agent-service-principal`;

-- 3. 데이터 테이블에는 SELECT만 부여 (INSERT/UPDATE/DELETE 절대 금지)
GRANT SELECT ON TABLE catalog.schema.orders
  TO `agent-service-principal`;

-- 4. 위험한 함수(삭제, 이메일 발송 등)는 별도 승인 프로세스 구축
-- → Human-in-the-loop 패턴으로 사용자 확인 후 실행

# Human-in-the-loop 패턴 예시
def cancel_order_with_confirmation(order_id: int) -> str:
    """주문을 취소합니다. 실행 전 사용자 확인이 필요합니다."""

    # 1단계: 주문 정보를 조회하여 사용자에게 보여줌
    order_info = get_order_details(order_id)

    # 2단계: 확인 메시지 반환 (실제 취소는 하지 않음)
    return (
        f"주문 {order_id} ({order_info['product_name']}, "
        f"{order_info['amount']}원)을 취소하시겠습니까? "
        f"확인하시면 '네, 취소해주세요'라고 말씀해주세요."
    )
    # 3단계: 사용자가 확인하면 별도 confirm_cancel_order 도구가 실행됨