[PYTHON/LANGCHAIN] 도구 호출시 사용자 승인 절차 추가하기

■ 도구 호출시 사용자 승인 절차를 추가하는 방법을 보여준다.

※ OPENAI_API_KEY 환경 변수 값은 .env 파일에 정의한다.

▶ main.py


import json

from dotenv                   import load_dotenv
from langchain_core.tools     import tool
from langchain_openai         import ChatOpenAI
from langchain_core.messages  import AIMessage
from typing                   import Dict
from typing                   import List

load_dotenv()

@tool
def countEmail(lastDayCount : int) -> int:
    """Dummy function to count number of e-mails. Returns 2 * last_n_days."""
    return lastDayCount * 2

@tool
def sendEmail(message : str, recipient : str) -> str:
    """Dummy function for sending an e-mail."""
    return f"Successfully sent email to {recipient}."

toolList = [countEmail, sendEmail]

chatOpenAI = ChatOpenAI(model = "gpt-4o-mini")

runnableBinding = chatOpenAI.bind_tools(toolList)

def callTool(aiMessage : AIMessage) -> List[Dict]:
    """Simple sequential tool calling helper."""
    toolMapDictionary = {tool.name : tool for tool in toolList}
    toolCallList = aiMessage.tool_calls.copy()
    for toolCall in toolCallList:
        toolCall["output"] = toolMapDictionary[toolCall["name"]].invoke(toolCall["args"])
    return toolCallList

class NotApproved(Exception):
    """Custom exception."""

def approve(aiMessage : AIMessage) -> AIMessage:
    """Responsible for passing through its input or raising an exception.

    Args:
        aiMessage : output from the chat model

    Returns:
        aiMessage : original output from the msg
    """

    toolCallListJSON = "\n\n".join(json.dumps(toolCall, indent = 2) for toolCall in aiMessage.tool_calls)
    inputString      = f"Do you approve of the following tool invocations\n\n{toolCallListJSON}\n\nAnything except 'Y'/'Yes' (case-insensitive) will be treated as a no.\n >>>"

    outputString = input(inputString)

    if outputString.lower() not in ("yes", "y"):
        raise NotApproved(f"Tool invocations not approved:\n\n{toolCallListJSON}")
    return aiMessage

runnableSequence = runnableBinding | approve | callTool

try:
    toolCallList = runnableSequence.invoke("how many emails did i get in the last 5 days?")

    print(toolCallList)
except NotApproved as e:
    print()
    print(e)

import json

from dotenv import load_dotenv

from langchain_core.tools import tool

from langchain_openai import ChatOpenAI

from langchain_core.messages import AIMessage

from typing import Dict

from typing import List

load_dotenv()

@tool

def countEmail(lastDayCount : int) -> int:

"""Dummy function to count number of e-mails. Returns 2 * last_n_days."""

return lastDayCount * 2

@tool

def sendEmail(message : str, recipient : str) -> str:

"""Dummy function for sending an e-mail."""

return f"Successfully sent email to {recipient}."

toolList = [countEmail, sendEmail]

chatOpenAI = ChatOpenAI(model = "gpt-4o-mini")

runnableBinding = chatOpenAI.bind_tools(toolList)

def callTool(aiMessage : AIMessage) -> List[Dict]:

"""Simple sequential tool calling helper."""

toolMapDictionary = {tool.name : tool for tool in toolList}

toolCallList = aiMessage.tool_calls.copy()

for toolCall in toolCallList:

toolCall["output"] = toolMapDictionary[toolCall["name"]].invoke(toolCall["args"])

return toolCallList

class NotApproved(Exception):

"""Custom exception."""

def approve(aiMessage : AIMessage) -> AIMessage:

"""Responsible for passing through its input or raising an exception.

Args:

aiMessage : output from the chat model

Returns:

aiMessage : original output from the msg

"""

toolCallListJSON = "\n\n".join(json.dumps(toolCall, indent = 2) for toolCall in aiMessage.tool_calls)

inputString = f"Do you approve of the following tool invocations\n\n{toolCallListJSON}\n\nAnything except 'Y'/'Yes' (case-insensitive) will be treated as a no.\n >>>"

outputString = input(inputString)

if outputString.lower() not in ("yes", "y"):

raise NotApproved(f"Tool invocations not approved:\n\n{toolCallListJSON}")

return aiMessage

runnableSequence = runnableBinding | approve | callTool

try:

toolCallList = runnableSequence.invoke("how many emails did i get in the last 5 days?")

print(toolCallList)

except NotApproved as e:

print()

print(e)

▶ requirements.txt


annotated-types==0.7.0
anyio==4.6.0
certifi==2024.8.30
charset-normalizer==3.3.2
colorama==0.4.6
distro==1.9.0
h11==0.14.0
httpcore==1.0.5
httpx==0.27.2
idna==3.10
jiter==0.5.0
jsonpatch==1.33
jsonpointer==3.0.0
langchain-core==0.3.5
langchain-openai==0.2.0
langsmith==0.1.125
openai==1.47.0
orjson==3.10.7
packaging==24.1
pydantic==2.9.2
pydantic_core==2.23.4
python-dotenv==1.0.1
PyYAML==6.0.2
regex==2024.9.11
requests==2.32.3
sniffio==1.3.1
tenacity==8.5.0
tiktoken==0.7.0
tqdm==4.66.5
typing_extensions==4.12.2
urllib3==2.2.3

annotated-types==0.7.0

anyio==4.6.0

certifi==2024.8.30

charset-normalizer==3.3.2

colorama==0.4.6

distro==1.9.0

h11==0.14.0

httpcore==1.0.5

httpx==0.27.2

idna==3.10

jiter==0.5.0

jsonpatch==1.33

jsonpointer==3.0.0

langchain-core==0.3.5

langchain-openai==0.2.0

langsmith==0.1.125

openai==1.47.0

orjson==3.10.7

packaging==24.1

pydantic==2.9.2

pydantic_core==2.23.4

python-dotenv==1.0.1

PyYAML==6.0.2

regex==2024.9.11

requests==2.32.3

sniffio==1.3.1

tenacity==8.5.0

tiktoken==0.7.0

tqdm==4.66.5

typing_extensions==4.12.2

urllib3==2.2.3

※ pip install python-dotenv langchain-openai 명령을 실행했다.