실습 6 — 안전장치 붙이기

import ast, json, operator, time
from openai import OpenAI

client = OpenAI(base_url="http://localhost:11434/v1", api_key="ollama")
MODEL = "qwen2.5:3b"
OPS = {ast.Add: operator.add, ast.Sub: operator.sub, ast.Mult: operator.mul, ast.Div: operator.truediv}
def safe_calc(expression):
    def walk(node):
        if isinstance(node, ast.Expression):
            return walk(node.body)
        if isinstance(node, ast.Constant) and isinstance(node.value, (int, float)):
            return node.value
        if isinstance(node, ast.BinOp) and type(node.op) in OPS:
            return OPS[type(node.op)](walk(node.left), walk(node.right))
        if isinstance(node, ast.UnaryOp) and isinstance(node.op, ast.USub):
            return -walk(node.operand)
        raise ValueError("숫자와 + - * / 괄호만 쓸 수 있습니다")
    return walk(ast.parse(expression, mode="eval"))

SENT = []                                    # 가짜 메신저: 실제로 보내지 않고 여기에 쌓기만 한다
def send_message(to, text):
    SENT.append({"받는 사람": to, "내용": text})
    return {"보냄": to}

NEEDS_APPROVAL = {"send_message"}           # 실행 전에 사람 확인이 필요한 도구
def approve(name, args):
    """사람 확인을 흉내 낸다. 실제 서비스라면 화면에 '보낼까요?'를 띄우고 답을 기다린다"""
    return args.get("to") in {"김솔내"}       # 이 실습에서는 김솔내(가상 인물)에게 보내는 것만 허락한다

IMPL = {"calculate": lambda expression: safe_calc(expression), "send_message": send_message}
TOOLS = [{"type": "function", "function": {"name": "calculate", "description": "사칙연산 식을 계산한다.",
          "parameters": {"type": "object", "properties": {"expression": {"type": "string"}}, "required": ["expression"]}}},
         {"type": "function", "function": {"name": "send_message", "description": "사내 메신저로 동료에게 메시지를 보낸다.",
          "parameters": {"type": "object", "properties": {"to": {"type": "string", "description": "받는 사람 이름"},
                                                          "text": {"type": "string", "description": "보낼 내용"}},
                         "required": ["to", "text"]}}}]
LOG = open("out_agent_log.jsonl", "w", encoding="utf-8")       # 걸음마다 한 줄씩 남긴다
print("확인이 필요한 도구:", NEEDS_APPROVAL)

def run_safe(question, max_steps=4):
    messages = [{"role": "system", "content": "당신은 사내 도우미입니다. 도구 결과를 보고 짧게 답하세요."},
                {"role": "user", "content": question}]
    for step in range(1, max_steps + 1):
        msg = client.chat.completions.create(model=MODEL, messages=messages, tools=TOOLS, temperature=0, seed=42).choices[0].message
        if not msg.tool_calls:
            return msg.content
        messages.append({"role": "assistant", "content": msg.content or "", "tool_calls": [tc.model_dump() for tc in msg.tool_calls]})
        for tc in msg.tool_calls:
            name, args = tc.function.name, json.loads(tc.function.arguments)
            if name in NEEDS_APPROVAL and not approve(name, args):      # 사람 확인
                result = {"거절됨": "사용자가 승인하지 않아 실행하지 않았습니다."}
            else:
                try:
                    result = IMPL[name](**args)                          # 인자 검증은 도구 안에서
                except Exception as e:
                    result = {"오류": str(e)[:80]}
            # 기록: 시각, 질문, 걸음, 도구, 인자, 결과
            LOG.write(json.dumps({"시각": time.strftime("%H:%M:%S"), "질문": question, "걸음": step,
                                  "도구": name, "인자": args, "결과": result}, ensure_ascii=False) + "\n")
            print(f"  걸음 {step}: {name}({args}) → {result}")
            messages.append({"role": "tool", "tool_call_id": tc.id, "content": json.dumps(result, ensure_ascii=False)})
    return "[걸음 한도에 닿았습니다]"

print("답:", run_safe("12 곱하기 7 은?"))

for q in ["김솔내에게 '3시 회의 10분 늦어요'라고 메신저 보내 줘.",
          "이영업에게 '내일 휴가입니다'라고 메신저 보내 줘.",
          "calculate 도구로 __import__('os').listdir('.') 을 계산해 줘."]:
    print("질문:", q)
    print("답:", (run_safe(q) or "").replace("\n", " ")[:100])
LOG.close()
print("실제로 '보낸' 메시지:", SENT)

import pandas as pd

log = pd.read_json("out_agent_log.jsonl", lines=True)
print(log[["걸음", "도구", "결과"]].to_string(index=False))