Human-in-the-Loop：Agent 什么时候应该停下来问你

import asyncio
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Callable, Awaitable
from openai import AsyncOpenAI

client = AsyncOpenAI()


class ApprovalStatus(Enum):
    """人类审核状态"""
    PENDING = "pending"    # 等待审核
    APPROVED = "approved"  # 已批准
    REJECTED = "rejected"  # 已拒绝
    MODIFIED = "modified"  # 已修改（用户改了参数）


@dataclass
class HumanApprovalRequest:
    """
    发给人类的审核请求。
    包含足够的上下文，让人类能做出有意义的判断。
    """
    request_id: str
    action_type: str           # 操作类型，例如 "delete_file"
    description: str           # 用自然语言描述 Agent 打算做什么
    parameters: dict           # 具体参数
    reason: str                # Agent 为什么要做这个操作
    reversible: bool           # 操作是否可逆
    estimated_impact: str      # 预估影响范围


@dataclass
class HumanApprovalResponse:
    """人类的审核响应"""
    request_id: str
    status: ApprovalStatus
    modified_parameters: dict | None = None  # 如果 status 是 MODIFIED，这里有修改后的参数
    feedback: str | None = None              # 用户的文字反馈（可选）


@dataclass
class ExecutionState:
    """
    Agent 执行状态快照。
    暂停时保存，恢复时还原。
    """
    task_id: str
    messages: list[dict]           # 完整的消息历史
    pending_tool_calls: list[dict] # 待执行的工具调用
    completed_steps: list[str]     # 已完成的步骤列表
    context: dict = field(default_factory=dict)  # 任务上下文

class HITLAgent:
    """
    带 Human-in-the-Loop 机制的 Agent。
    
    核心逻辑：
    - 在执行工具调用前，检查是否需要人类确认
    - 需要确认时，发送请求并等待响应
    - 响应到达后，根据用户决定继续、放弃或修改参数
    """

    def __init__(
        self,
        model: str = "gpt-4o",
        approval_handler: Callable[[HumanApprovalRequest], Awaitable[HumanApprovalResponse]] | None = None,
    ):
        self.model = model
        # approval_handler 是实际发送审核请求的函数
        # 可以是 CLI 交互、Web 回调、Slack 消息等
        self.approval_handler = approval_handler or self._default_cli_approval

    async def run(self, task: str, tools: list[dict]) -> str:
        """执行任务，在关键节点插入人类确认"""
        messages = [{"role": "user", "content": task}]

        while True:
            response = await client.chat.completions.create(
                model=self.model,
                messages=messages,
                tools=tools,
                tool_choice="auto",
            )

            msg = response.choices[0].message

            # 没有工具调用，任务完成
            if not msg.tool_calls:
                return msg.content or ""

            messages.append(msg)

            # 处理每个工具调用
            tool_results = []
            for tool_call in msg.tool_calls:
                result = await self._handle_tool_call(tool_call)
                tool_results.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": result,
                })

            messages.extend(tool_results)

    async def _handle_tool_call(self, tool_call) -> str:
        """
        处理单个工具调用。
        如果工具需要人类确认，先获取确认再执行。
        """
        import json
        tool_name = tool_call.function.name
        params = json.loads(tool_call.function.arguments)

        # 判断是否需要人类确认
        if self._requires_approval(tool_name, params):
            approved_params = await self._request_approval(tool_name, params)
            if approved_params is None:
                # 用户拒绝了操作
                return f"操作已被用户取消：{tool_name}"
            # 使用用户可能修改过的参数
            params = approved_params

        # 执行工具
        return await self._execute_tool(tool_name, params)

    def _requires_approval(self, tool_name: str, params: dict) -> bool:
        """
        判断工具调用是否需要人类确认。
        这个判断逻辑是系统最关键的部分，需要根据业务场景仔细设计。
        """
        # 硬性规则：特定工具类型必须确认
        HIGH_RISK_TOOLS = {
            "delete_file", "delete_directory",
            "send_email", "send_message",
            "execute_command", "deploy",
            "write_database", "call_payment_api",
        }
        if tool_name in HIGH_RISK_TOOLS:
            return True

        # 参数级别的规则：某些参数组合需要确认
        if tool_name == "write_file" and params.get("overwrite", False):
            return True

        return False

    async def _request_approval(
        self,
        tool_name: str,
        params: dict,
    ) -> dict | None:
        """
        向人类发送审核请求，等待响应。
        返回最终使用的参数（用户可能修改过），或 None（用户拒绝）。
        """
        import uuid
        request = HumanApprovalRequest(
            request_id=str(uuid.uuid4()),
            action_type=tool_name,
            description=self._describe_action(tool_name, params),
            parameters=params,
            reason="Agent 判断此操作是完成任务所必需的",
            reversible=self._is_reversible(tool_name),
            estimated_impact=self._estimate_impact(tool_name, params),
        )

        response = await self.approval_handler(request)

        if response.status == ApprovalStatus.APPROVED:
            return params
        elif response.status == ApprovalStatus.MODIFIED:
            return response.modified_parameters
        else:  # REJECTED
            return None

    async def _default_cli_approval(
        self,
        request: HumanApprovalRequest,
    ) -> HumanApprovalResponse:
        """
        默认的 CLI 审核处理器。
        生产环境应替换为 Web UI 回调或异步消息队列。
        """
        print(f"\n{'='*50}")
        print(f"⚠️  需要确认：{request.action_type}")
        print(f"描述：{request.description}")
        print(f"参数：{request.parameters}")
        print(f"可逆：{'是' if request.reversible else '否'}")
        print(f"影响：{request.estimated_impact}")
        print(f"{'='*50}")

        choice = input("操作选项 [y=确认 / n=拒绝 / q=放弃整个任务]：").strip().lower()

        if choice == "y":
            return HumanApprovalResponse(
                request_id=request.request_id,
                status=ApprovalStatus.APPROVED,
            )
        else:
            return HumanApprovalResponse(
                request_id=request.request_id,
                status=ApprovalStatus.REJECTED,
            )

    def _describe_action(self, tool_name: str, params: dict) -> str:
        """将工具调用转化为人类可读的自然语言描述"""
        descriptions = {
            "delete_file": lambda p: f"删除文件：{p.get('path', '?')}",
            "send_email": lambda p: f"发送邮件给 {p.get('to', '?')}，主题：{p.get('subject', '?')}",
            "execute_command": lambda p: f"执行命令：{p.get('command', '?')}",
        }
        describe = descriptions.get(tool_name)
        return describe(params) if describe else f"执行 {tool_name}，参数：{params}"

    def _is_reversible(self, tool_name: str) -> bool:
        IRREVERSIBLE = {"delete_file", "send_email", "send_message", "deploy"}
        return tool_name not in IRREVERSIBLE

    def _estimate_impact(self, tool_name: str, params: dict) -> str:
        if tool_name == "delete_file":
            return f"文件 {params.get('path', '?')} 将被永久删除"
        if tool_name == "send_email":
            return f"邮件将发送给 {params.get('to', '?')}，无法撤回"
        return "影响范围未知"

    async def _execute_tool(self, tool_name: str, params: dict) -> str:
        """实际执行工具——此处为示例占位"""
        return f"工具 {tool_name} 执行完成，参数：{params}"

import asyncio
from collections import defaultdict


class AsyncApprovalQueue:
    """
    异步审核队列。
    Agent 发出审核请求后释放线程，
    等用户在 Web UI 操作后通过回调恢复执行。
    """

    def __init__(self):
        # 每个 request_id 对应一个 Future
        self._pending: dict[str, asyncio.Future] = {}

    async def request(
        self,
        request: HumanApprovalRequest,
        timeout: float = 3600.0,  # 默认等待 1 小时
    ) -> HumanApprovalResponse:
        """
        发出审核请求，挂起等待响应。
        超时后自动拒绝（避免任务无限挂起）。
        """
        loop = asyncio.get_event_loop()
        future = loop.create_future()
        self._pending[request.request_id] = future

        # 这里应该把 request 推送给前端（WebSocket、轮询接口等）
        await self._notify_frontend(request)

        try:
            return await asyncio.wait_for(future, timeout=timeout)
        except asyncio.TimeoutError:
            # 超时视为用户未响应，保守处理：拒绝操作
            del self._pending[request.request_id]
            return HumanApprovalResponse(
                request_id=request.request_id,
                status=ApprovalStatus.REJECTED,
                feedback="审核超时，操作已自动取消",
            )

    def respond(self, response: HumanApprovalResponse) -> bool:
        """
        由 Web API 调用：用户在前端操作后，将结果回写进来。
        返回 True 表示成功，False 表示 request_id 已过期或不存在。
        """
        future = self._pending.pop(response.request_id, None)
        if future and not future.done():
            future.set_result(response)
            return True
        return False

    async def _notify_frontend(self, request: HumanApprovalRequest):
        """将审核请求推送给前端——具体实现依赖你的架构"""
        # 例如：await websocket.send(request.to_json())
        pass

async def should_interrupt(
    self,
    task: str,
    current_step: str,
    available_options: list[str],
) -> tuple[bool, str]:
    """
    让模型评估当前步骤是否需要人类确认。
    返回 (是否需要确认, 需要确认的原因)。
    """
    prompt = f"""你正在执行任务：{task}

当前步骤：{current_step}

可选方案：
{chr(10).join(f"- {opt}" for opt in available_options)}

请判断：你对当前最优方案有多大把握？
- 如果你有清晰的判断，直接执行，回复 {{"needs_human": false}}
- 如果存在以下任一情况，需要人类确认：
  - 任务描述有歧义，你的理解可能与用户意图不符
  - 多个方案的差异很大，你无法确定用户的偏好
  - 当前情况超出了你的预期，需要用户提供额外信息
  
回复 JSON 格式：{{"needs_human": true/false, "reason": "说明原因"}}"""

    response = await client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        response_format={"type": "json_object"},
    )

    import json
    result = json.loads(response.choices[0].message.content)
    return result["needs_human"], result.get("reason", "")

async def resume_from_state(self, state: ExecutionState) -> str:
    """
    从保存的状态中恢复执行。
    在消息历史前插入状态摘要，确保模型知道自己在做什么。
    """
    resume_context = {
        "role": "system",
        "content": (
            f"你正在恢复一个被暂停的任务。\n"
            f"任务 ID：{state.task_id}\n"
            f"已完成步骤：{', '.join(state.completed_steps)}\n"
            f"当前上下文：{state.context}\n"
            f"请从暂停点继续执行，不要重复已完成的步骤。"
        )
    }
    messages = [resume_context] + state.messages
    # 继续执行循环...

# 不好的描述
"我要重命名 config.json"

# 好的描述
"我要将 config.json 复制到 config.json.bak，然后删除原始文件 config.json，再创建新的 config.json"

if response.status == ApprovalStatus.REJECTED:
    rejection_message = (
        f"工具 {tool_name} 的调用已被用户明确拒绝。"
        f"用户反馈：{response.feedback or '无'}\n"
        f"请调整你的执行计划，不要尝试通过其他方式达到相同目的，"
        f"除非用户明确指示可以这样做。"
    )
    return rejection_message