Human-in-the-Loop 도구 확인
Human-in-the-Loop 도구 확인 (Human-in-the-Loop Tool Confirmation)
도구 호출에 승인 흐름을 추가해서, 실행 전에 사용자가 작업을 검토하고 확인하거나 거부할 수 있게 하는 방법을 다루는 쿡북이에요.
출처: 문서
본문
도구 호출에 승인 흐름(approval flows)을 추가해서, 실행 전에 사용자가 작업을 검토하고 확인하거나 거부할 수 있게 해 봅시다.
API 상태 (API status): 이 쿡북은 client.beta.conversations를 사용해요. 이는 베타(beta) 엔드포인트로 변경될 수 있어요.
사전 준비사항 (Prerequisites)
설치 (Install)
# Python
pip install mistralai
# or with uv
uv add mistralai
필요한 환경 변수 (Required environment variables)
이 쿡북을 완료하려면 Mistral API 키가 필요해요. Studio에서 API keys 섹션으로 이동해서 새 API 키를 만들어요.
프로젝트 루트에 .env를 만들고 Mistral API 키를 추가해요.
MISTRAL_API_KEY=your-mistral-api-key
개념 (Concepts)
RunContext.run_async 루프로 두 가지 지연(deferral) 흐름을 사용할 수 있어요.
- 클라이언트 측(client-side): 로컬 MCP 클라이언트 및 함수를
register_mcp_client또는register_func로 등록. - 서버 측(server-side): 원격 Mistral 커넥터(gmail 등).
run_async 루프는 지연된 도구 호출을 만나면 스스로 중단하는 책임을 져요. 지연된 도구 호출은 로컬 함수에 의해서도, 서버 측 이벤트(confirmation_status: "pending"인 FunctionCallEntry)에 의해서도 나타날 수 있어요.
구성 (Configuration)
확인 요구 동작을 구성하려면 다음 도구 선언 구조를 사용해요.
tools=[
{
"type": "connector",
"connector_id": "gmail",
"tool_configuration": {
"include": ["gmail_search"],
"exclude": ["gmail_send"], # mutually exclusive with include
"requires_confirmation": ["gmail_search"],
},
},
{
"type": "web_search_premium",
"tool_configuration": {
"requires_confirmation": ["web_search", "news_search"],
},
},
]
루프 패턴 (Loop pattern)
아래 레시피들은 while True 루프를 사용해서 DeferredToolCallsException을 잡고, 승인을 요청하고, 한 스크립트 안에서 대화를 재개해요. 데모와 CLI 도구에 편리해요.
프로덕션에서는 지연과 재개가 보통 별도의 컨텍스트에서 일어나요 — 예를 들어 백엔드가 지연을 잡아 보류 중인 도구 호출을 프론트엔드로 보내 사용자 승인을 받고, 프론트엔드가 응답하면 대화를 재개하죠. Recipe 3 (Serialize and Resume)이 이 패턴을 보여줘요.
레시피 (Recipes)
1. 확인이 있는 로컬 함수 (Local Functions with Confirmation)
언제 쓰나 (When to use):
- 모델이 어떤 배선 없이 직접 실행하기를 원하는 로컬 Python 함수가 있을 때
- 일부는 안전하고(예: 읽기 전용 조회) 자동 실행되어 에이전트 루프를 계속 이어가야 할 때
- 일부는 사람의 승인이 필요하고(예: 쓰기 연산, 예약 등), 실행 전에 승인이 필요할 때
import asyncio
import os
import random
from mistralai import Mistral
from mistralai.extra.run.context import RunContext
from mistralai.extra.exceptions import DeferredToolCallsException
MODEL = "mistral-large-latest"
def get_weather(city: str) -> str:
"""Get the current weather for a city."""
temp = random.randint(10, 30)
conditions = random.choice(["sunny", "cloudy", "partly cloudy"])
return f"The weather in {city} is {conditions}, {temp}C"
def book_flight(destination: str, date: str) -> str:
"""Book a flight to a destination."""
return f"Flight booked to {destination} on {date}. Confirmation: FL-{random.randint(10000, 99999)}"
def request_approval(dc) -> bool:
print(f"\n[APPROVAL REQUIRED] {dc.tool_name}")
print(f" Arguments: {dc.arguments}")
return input(" Approve? (y/n): ").strip().lower() == "y"
async def main():
client = Mistral(api_key=os.environ["MISTRAL_API_KEY"])
conversation_id = None
pending_inputs = [
{"role": "user", "content": "I need a vacation somewhere warm next Friday. Can you help?"}
]
while True:
async with RunContext(model=MODEL) as run_ctx:
run_ctx.conversation_id = conversation_id
run_ctx.register_func(get_weather, requires_confirmation=False)
run_ctx.register_func(book_flight, requires_confirmation=True)
try:
result = await client.beta.conversations.run_async(
run_ctx=run_ctx,
inputs=pending_inputs,
instructions="You are a travel assistant. Available destinations are: Guingamp, Aurillac, Brive-la-Gaillarde, Rodez, and Millau. Check the weather and book a flight to the warmest one. Do not ask for confirmation, just book it.",
)
print(f"\nFinal response: {result.output_entries}")
break
except DeferredToolCallsException as deferred:
conversation_id = deferred.conversation_id
pending_inputs = [
dc.confirm() if request_approval(dc) else dc.reject("Denied by user")
for dc in deferred.deferred_calls
]
asyncio.run(main())
작동 방식 (How it works):
requires_confirmation=False로 등록한 함수는 모델이 호출하면 자동 실행돼요.requires_confirmation=True로 등록한 함수는 실행을 멈추고 대신DeferredToolCallsException을 발생시켜요.- 이 예외는 보류 중인 도구 호출을 담고 있어요. 각각에
dc.confirm()또는dc.reject()를 호출한 다음, 이를inputs로 다시 전달해서 대화를 재개해요.
2. 확인이 있는 커넥터 (Gmail, Connector with Confirmation)
언제 쓰나 (When to use):
- 모델에게 원격 Mistral 커넥터(예: Gmail) 접근을 주고 싶을 때
- 일부 연산에 사람의 승인을 원할 때
사전 준비사항 (Prereqs): 유효한 Google OAuth2 토큰(GMAIL_OAUTH_TOKEN 환경 변수).
import asyncio
import os
from mistralai import Mistral
from mistralai.extra.exceptions import DeferredToolCallsException
from mistralai.extra.run.context import RunContext
MODEL = "mistral-large-latest"
def request_approval(dc) -> bool:
print(f"\n[APPROVAL REQUIRED] {dc.tool_name}")
print(f" Arguments: {dc.arguments}")
return input(" Approve? (y/n): ").strip().lower() == "y"
async def main():
client = Mistral(api_key=os.environ["MISTRAL_API_KEY"])
conversation_id = None
pending_inputs = [
{"role": "user", "content": "Summarize my latest emails from Gmail."}
]
while True:
async with RunContext(model=MODEL) as run_ctx:
run_ctx.conversation_id = conversation_id
try:
result = await client.beta.conversations.run_async(
run_ctx=run_ctx,
inputs=pending_inputs,
instructions="You are a helpful assistant. Use the Gmail connector to access the user's emails.",
tools=[
{
"type": "connector",
"connector_id": "gmail",
"authorization": {
"type": "oauth2-token",
"value": os.environ["GMAIL_OAUTH_TOKEN"],
},
"tool_configuration": {
"requires_confirmation": ["gmail_search"],
},
},
],
)
for entry in result.output_entries:
if hasattr(entry, "content"):
print(f"\n[{entry.type}] {entry.content}")
else:
print(f"\n[{entry.type}] {entry.name}({entry.arguments})")
break
except DeferredToolCallsException as deferred:
conversation_id = deferred.conversation_id
pending_inputs = [
dc.confirm() if request_approval(dc) else dc.reject("Denied by user")
for dc in deferred.deferred_calls
]
asyncio.run(main())
작동 방식 (How it works):
requires_confirmation에 나열된 도구 이름은 즉시 실행되지 않고 서버 측에서 일시 중지돼요.run_async가 일시 중지된 도구를 감지하고DeferredToolCallsException을 발생시켜요.dc.confirm()을 호출하면 실행을 허용하고,dc.reject()를 호출하면 거부해요. 이 결정은 대화가 재개될 때 서버로 다시 보내져요.
3. 무상태/API 친화적 — 직렬화 및 재개 (Stateless / API-Friendly — Serialize and Resume)
언제 쓰나 (When to use):
- 승인 단계가 대화를 시작한 것과 다른 프로세스나 서비스에서 일어날 때
- 도구 호출, 실행, 재개 요청의 직렬화가 필요할 때
- 실제 API 경계를 시뮬레이션하기 위해 두 개의 스크립트로 나눌 때(예: 백엔드가 지연 상태를 프론트엔드로 반환, 프론트엔드가 승인을 다시 보냄)
스크립트 1: 대화 시작, 지연 잡기, 직렬화 (Start the conversation, catch the deferral, serialize it)
import asyncio
import json
import os
from mistralai import Mistral
from mistralai.extra.run.context import RunContext
from mistralai.extra.exceptions import DeferredToolCallsException
def book_flight(destination: str, date: str) -> str:
"""Book a flight to a destination."""
return f"Flight booked to {destination} on {date}"
async def main():
client = Mistral(api_key=os.environ["MISTRAL_API_KEY"])
async with RunContext(model="mistral-large-latest") as run_ctx:
run_ctx.register_func(book_flight, requires_confirmation=True)
try:
result = await client.beta.conversations.run_async(
run_ctx=run_ctx,
inputs=[{"role": "user", "content": "Book me a flight to Paris next Friday."}],
instructions="You are a travel assistant. Book the flight directly.",
)
print("No confirmation needed:", result.output_as_text)
except DeferredToolCallsException as deferred:
state = deferred.to_dict()
serialized = json.dumps(state)
print("Deferred state (send this to your frontend / store it):")
print(serialized)
asyncio.run(main())
스크립트 2: 승인 받기, 역직렬화, 재개 (Receive approvals, deserialize, and resume)
import asyncio
import json
import os
from mistralai import Mistral
from mistralai.extra.run.context import RunContext
from mistralai.extra.exceptions import DeferredToolCallsException, DeferredToolCallEntry
def book_flight(destination: str, date: str) -> str:
"""Book a flight to a destination."""
return f"Flight booked to {destination} on {date}"
async def main():
# In a real app: receive this from the frontend / load from DB
serialized = os.environ["DEFERRED_STATE"] # the JSON string from Script 1
state = json.loads(serialized)
# Reconstruct the exception from the serialized state
deferred = DeferredToolCallsException.from_dict(state)
# Build confirmations (in a real app, the frontend tells you which to approve/reject)
pending_inputs = []
for dc in deferred.deferred_calls:
print(f"Tool: {dc.tool_name}, Args: {dc.arguments}")
pending_inputs.append(dc.confirm())
# Include any already-executed results
pending_inputs = list(deferred.executed_results) + pending_inputs
# Resume the conversation
client = Mistral(api_key=os.environ["MISTRAL_API_KEY"])
async with RunContext(model="mistral-large-latest") as run_ctx:
run_ctx.conversation_id = deferred.conversation_id
run_ctx.register_func(book_flight, requires_confirmation=True)
result = await client.beta.conversations.run_async(
run_ctx=run_ctx,
inputs=pending_inputs,
instructions="You are a travel assistant. Book the flight directly.",
)
print("Final response:", result.output_as_text)
asyncio.run(main())
작동 방식 (How it works):
deferred.to_dict()는 전체 지연 상태(대화 ID, 보류 중인 도구 호출, 이미 실행된 결과)를 저장하거나 와이어로 보낼 수 있는 평범한 dict로 직렬화해요.- 별도 프로세스에서
DeferredToolCallsException.from_dict(state)가 그 상태를 재구성해요. 거기서 호출을 확인하거나 거부하고 평소처럼 대화를 재개해요. - 재개 프로세스는 승인 후 실행될 수 있도록 동일한 로컬 함수를
register_func로 다시 등록해야 해요.
요약 (Summary)
이 쿡북은 Mistral 대화에서 도구 호출에 human-in-the-loop 승인 흐름을 추가하는 방법을 다뤘어요. 실행 전에 보류 중인 도구 호출을 가로채고, 확인을 요청하고, 승인하거나 거부하는 방식이에요. 지연된 대화를 직렬화해서 별도 프로세스에서 재개하는 방법도 보여줬어요.
이 쿡북이 다루는 내용 (What this cookbook covers):
- 로컬 함수로 도구 확인
- Connectors(Gmail)로 도구 확인
- 무상태 승인 흐름: 지연을 직렬화하고, 전송하고, 별도 프로세스에서 재개하기
사용한 Mistral 기능 (Mistral features used):
- Conversations API (beta)
- Connectors (beta)
기타 서비스 (Other services):
- Gmail — OAuth2 인증 Connector