실습 2 — 도구 실행기 만들기

import pandas as pd
from langchain_core.tools import tool

products = pd.read_csv("trip_products.csv")


@tool
def find_products(region: str = "", max_price: int = 0) -> str:
    """구름다리여행 상품을 지역과 가격(어른 1명, 만 원 상한)으로 찾는다. 0 이면 가격을 따지지 않는다."""
    df = products
    if region:
        df = df[df["지역"] == region]
    if max_price > 0:
        df = df[df["가격"] <= max_price]
    if df.empty:
        return "조건에 맞는 상품이 없습니다."
    return "\n".join(
        f"{r.상품코드} {r.상품명} | {r.가격}만 원 | 남은 자리 {r.남은자리} | 출발 {r.출발일}" for r in df.itertuples()
    )


@tool
def calc_price(code: str, adults: int, children: int = 0) -> str:
    """상품코드와 인원으로 총 여행 요금(만 원)을 계산한다. 아동 요금은 어른의 80% 이다."""
    rows = products[products["상품코드"] == code]
    if rows.empty:
        return f"없는 상품코드입니다: {code}"
    row = rows.iloc[0]
    total = round(row["가격"] * adults + row["가격"] * 0.8 * children, 1)
    return f"{code} {row['상품명']} · 어른 {adults}명, 아동 {children}명 · {float(total):g}만 원"


# 이름으로 도구를 찾을 수 있게 사전으로 모은다
TOOLS = {t.name: t for t in [find_products, calc_price]}
print("도구 상자:", list(TOOLS))

from langchain_core.messages import AIMessage, HumanMessage

question = HumanMessage("부산 상품을 보여 주고, P04 에 어른 2명이면 얼마인지도 알려 주세요.")
# 모델이 한 번에 두 도구를 부르겠다고 답했다고 보고 손으로 만든 메시지
ai = AIMessage(content="", tool_calls=[
    {"name": "find_products", "args": {"region": "부산"}, "id": "call_1"},
    {"name": "calc_price", "args": {"code": "P04", "adults": 2}, "id": "call_2"},
])

print("질문:", question.content)
for call in ai.tool_calls:
    print(call["id"], call["name"], call["args"])

from langchain_core.messages import ToolMessage


def run_tool_calls(ai_message):
    """AIMessage 의 tool_calls 를 하나씩 실행해 ToolMessage 목록으로 돌려준다."""
    results = []
    for call in ai_message.tool_calls:
        name, args, call_id = call["name"], call["args"], call["id"]
        if name not in TOOLS:
            # 없는 도구: 멈추지 않고 오류를 결과로 돌려준다(모델이 다시 고를 수 있게)
            content, status = f"오류: {name} 라는 도구는 없습니다. 쓸 수 있는 도구: {', '.join(TOOLS)}", "error"
        else:
            try:
                content, status = TOOLS[name].invoke(args), "success"
            except Exception as e:
                # 인자가 틀린 경우(예: 숫자 자리에 글자): 오류 종류와 첫 줄만 돌려준다
                content, status = f"오류: {type(e).__name__} - {str(e).splitlines()[0]}", "error"
        # tool_call_id 에 요청 번호를 그대로 적어 짝을 맞춘다. status 는 성공·오류 표시
        results.append(ToolMessage(content=content, tool_call_id=call_id, name=name, status=status))
    return results


for msg in run_tool_calls(ai):
    print(f"[{msg.tool_call_id} {msg.name} {msg.status}]")
    print(msg.content)

# 모델이 틀리게 부른 두 가지 경우
bad = AIMessage(content="", tool_calls=[
    {"name": "book_product", "args": {"code": "P04"}, "id": "call_3"},
    {"name": "calc_price", "args": {"code": "P04", "adults": "둘"}, "id": "call_4"},
])
for msg in run_tool_calls(bad):
    print(f"[{msg.tool_call_id} {msg.name} {msg.status}] {msg.content}")

import json
from pathlib import Path

# 둘째로 모델에 보낼 메시지 목록: 질문, 도구 호출 요청, 결과 둘
messages = [question, ai] + run_tool_calls(ai)
print("메시지:", [m.type for m in messages])

log = []
for m in messages:
    item = {"type": m.type, "content": m.content}
    if m.type == "ai":
        item["tool_calls"] = [{"name": c["name"], "args": c["args"], "id": c["id"]} for c in m.tool_calls]
    if m.type == "tool":
        item["tool_call_id"] = m.tool_call_id
    log.append(item)

Path("out_ch05_tool_log.json").write_text(json.dumps(log, ensure_ascii=False, indent=2), encoding="utf-8")
print("저장: out_ch05_tool_log.json")