{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 人机协同环节：预执行关卡、风险分层和审计日志\n",
    "\n",
    "本课的 README 中通过一个简短示例介绍了人机协同，该示例在代理已生成响应后，要求用户选择 `APPROVE` 或 `REJECT`。这种模式是一个很好的起点，但实际的生产环境中，人机协同通常需要另外三个方面：\n",
    "\n",
    "1. 一个 <strong>预执行关卡</strong>，在代理执行高风险步骤之前运行，从而控制成本、不可逆性和延迟。\n",
    "2. <strong>风险分层</strong>，使低风险动作自动执行，中风险动作批量审批，只有高风险动作真正需要人工阻断。\n",
    "3. 一个 <strong>审计日志加修订循环</strong>，每个关卡决策都记录为 JSONL 格式，拒绝时用结构化理由重新提示代理，而不仅仅是打印 `Revising...`。\n",
    "\n",
    "本笔记本在与 `06-system-message-framework.ipynb` 相同的基本原语上构建这些功能。它可以在 `DEMO_MODE = True` 下端到端运行（无需交互输入），或者在 `DEMO_MODE = False` 采用真实的 `input()` 提示。注意：在 DEMO_MODE 下第三个目标的重试是预设的脚本以便机制端到端可见。真实的修订驱动重分类需要 `DEMO_MODE = False` 并由操作员执行。\n",
    "\n",
    "**不在本课范围（在其他课程中讲解）：** 认证与访问控制（第06课 README 的威胁2）、工具调用中间件（第14课 MAF 深入解析）、多代理辩论模式。\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import json\n",
    "import os\n",
    "from datetime import datetime, timezone\n",
    "from pathlib import Path\n",
    "\n",
    "from dotenv import load_dotenv\n",
    "from azure.identity import DefaultAzureCredential, get_bearer_token_provider\n",
    "from openai import OpenAI\n",
    "\n",
    "load_dotenv()\n",
    "\n",
    "DEMO_MODE = True  # set False to use real input() prompts\n",
    "\n",
    "# Per-run unique log filename so demo runs don't overwrite each other and\n",
    "# the notebook doesn't touch any pre-existing gate_log.jsonl in the working\n",
    "# directory.\n",
    "GATE_LOG_PATH = Path(\n",
    "    f\"gate_log_{datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%SZ')}.jsonl\"\n",
    ")\n",
    "\n",
    "# This notebook uses the Azure OpenAI Responses API via the stable /openai/v1/ endpoint.\n",
    "# GitHub Models is deprecated (retiring July 2026) and does not support the Responses API.\n",
    "endpoint = os.environ.get(\"AZURE_OPENAI_ENDPOINT\", \"\")\n",
    "if not endpoint:\n",
    "    raise RuntimeError(\n",
    "        \"AZURE_OPENAI_ENDPOINT environment variable is not set. This notebook needs \"\n",
    "        \"an Azure OpenAI resource with a model deployment that supports the Responses \"\n",
    "        \"API. Set AZURE_OPENAI_ENDPOINT and AZURE_OPENAI_DEPLOYMENT in \"\n",
    "        \"your environment or a local .env file, then run `az login`.\"\n",
    "    )\n",
    "\n",
    "deployment = os.environ[\"AZURE_OPENAI_DEPLOYMENT\"]\n",
    "\n",
    "# Authenticate with Entra ID (run `az login` first). No api_version is needed.\n",
    "token_provider = get_bearer_token_provider(\n",
    "    DefaultAzureCredential(),\n",
    "    \"https://cognitiveservices.azure.com/.default\",\n",
    ")\n",
    "\n",
    "client = OpenAI(\n",
    "    base_url=f\"{endpoint.rstrip('/')}/openai/v1/\",\n",
    "    api_key=token_provider,\n",
    ")\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 模式 1：预先操作门控\n",
    "\n",
    "README 中的 HITL 代码片段先调用代理，然后请求用户批准输出。这是一个<strong>后操作</strong>流程。代理已经执行，因此 LLM 调用费用已支付，任何副作用（发送的邮件、写入的数据库行、发布的评论）也已经发生。\n",
    "\n",
    "<strong>预先操作</strong>流程则是在代理运行风险步骤之前插入门控。代理提出操作，门控决定是否执行，只有在批准后副作用才会发生。\n",
    "\n",
    "| 方面 | 后操作审批（README 代码片段） | 预先操作门控（此笔记本） |\n",
    "|---|---|---|\n",
    "| 何时运行审批？ | 代理已生成输出后 | 在任何副作用执行前 |\n",
    "| 拒绝时的 LLM 费用 | 已支付 | 仅为提议支付，不为操作支付 |\n",
    "| 不可逆的副作用 | 可能（操作已发生） | 已防止 |\n",
    "| 审计清晰度 | 审批是打印语句 | 审批是带有时间戳、操作、原因的 JSON 记录 |\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def gate_action(action_description: str, risk_tier: str, attempt: int = 0) -> dict:\n",
    "    \"\"\"Run a single pre-action gate.\n",
    "\n",
    "    Returns a decision dict with keys: decision, reason, ts.\n",
    "    Decision is one of: approve, deny, escalate.\n",
    "    Safe default on EOF or unexpected input is deny.\n",
    "\n",
    "    DEMO_MODE behavior: high-risk actions are denied on attempt 0 and\n",
    "    auto-approved on attempt >= 1. This is scripted approval to show the\n",
    "    loop mechanics (deny -> retry -> approve). It is NOT revision-driven\n",
    "    re-classification. Real revision-driven re-classification requires\n",
    "    DEMO_MODE=False and a human operator who evaluates the revised\n",
    "    proposal on its own merits.\n",
    "    \"\"\"\n",
    "    print(f\"[gate] proposed action ({risk_tier}, attempt={attempt}): {action_description}\")\n",
    "\n",
    "    if DEMO_MODE:\n",
    "        if risk_tier == \"high\":\n",
    "            decision = \"approve\" if attempt >= 1 else \"deny\"\n",
    "            reason = (\n",
    "                \"DEMO_MODE: scripted approval on retry to show loop mechanics\"\n",
    "                if attempt >= 1\n",
    "                else \"DEMO_MODE: high risk denied on first attempt\"\n",
    "            )\n",
    "        else:\n",
    "            decision = \"approve\"\n",
    "            reason = f\"DEMO_MODE canned response for tier={risk_tier}\"\n",
    "    else:\n",
    "        try:\n",
    "            raw = input(\"[gate] approve / deny / escalate? \").strip().lower()\n",
    "        except EOFError:\n",
    "            raw = \"\"\n",
    "        if raw in {\"approve\", \"deny\", \"escalate\"}:\n",
    "            decision, reason = raw, \"operator input\"\n",
    "        elif raw == \"\":\n",
    "            decision, reason = \"deny\", \"no input received, defaulted to deny\"\n",
    "        else:\n",
    "            decision, reason = \"deny\", f\"invalid input {raw!r}, defaulted to deny\"\n",
    "\n",
    "    return {\n",
    "        \"decision\": decision,\n",
    "        \"reason\": reason,\n",
    "        \"action\": action_description,\n",
    "        \"risk_tier\": risk_tier,\n",
    "        \"ts\": datetime.now(timezone.utc).isoformat(),\n",
    "    }\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 模式 2：风险分级\n",
    "\n",
    "并非所有操作都需要人工批准。只读查询公共 API 的风险与发送客户邮件的风险不同。将两者一视同仁会浪费操作人员的注意力并减慢代理速度。\n",
    "\n",
    "一个简单的三层模型：\n",
    "\n",
    "| 层级 | 示例 | 审批流程 |\n",
    "|---|---|---|\n",
    "| `低`（只读） | 搜索知识库，查询航班选项，抓取公共网页 | 自动执行，记录以供审计 |\n",
    "| `中`（低成本变更） | 缓存结果，计数器递增，安排提醒 | 自动执行，但每日批量复审 |\n",
    "| `高`（面向外部或不可逆） | 发送电子邮件，扣款，发布到公共频道 | 阻塞，需人工审批 |\n",
    "\n",
    "这是一种分级方法。生产系统通常使用更细粒度的层级（例如，AWS IAM 权限级别、基于角色的访问层级）。下面的三层版本是适用于混合只读和副作用操作的代理的最小实用版本。\n",
    "\n",
    "下面的分类器使用关键词启发式，确保演示保持确定性和低成本。在生产系统中，您会用训练好的分类器或策略引擎替代它。\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "LOW_RISK_KEYWORDS = {\n",
    "    \"look\", \"lookup\", \"search\", \"fetch\", \"read\", \"query\", \"view\",\n",
    "    \"get\", \"list\", \"weather\", \"summarize\",\n",
    "}\n",
    "HIGH_RISK_KEYWORDS = {\n",
    "    \"send\", \"email\", \"post\", \"publish\", \"charge\", \"pay\", \"transfer\",\n",
    "    \"delete\", \"drop\", \"cancel\", \"refund\",\n",
    "}\n",
    "MEDIUM_RISK_KEYWORDS = {\n",
    "    \"cache\", \"schedule\", \"reminder\", \"book\", \"reserve\", \"update\",\n",
    "    \"increment\", \"log\",\n",
    "}\n",
    "\n",
    "AUTO_APPROVE_REASONS = {\n",
    "    \"low\": \"auto-approved (low risk)\",\n",
    "    \"medium\": \"auto-approved (medium risk, queued for batched review)\",\n",
    "}\n",
    "\n",
    "\n",
    "def classify_risk(action: str) -> str:\n",
    "    \"\"\"Classify an action string into one of: low, medium, high.\n",
    "\n",
    "    Keyword-based heuristic. Checks high-risk first (most severe), then\n",
    "    low-risk explicit reads, then medium-risk mutations. Unrecognized\n",
    "    actions default to medium, not low.\n",
    "\n",
    "    Default for unrecognized actions is 'medium', not 'low'. A read-only\n",
    "    keyword set will always have blind spots, and the parent README's\n",
    "    threat list (critical-system access, knowledge-base poisoning,\n",
    "    cascading errors) all involve cases an action-name alone cannot rule\n",
    "    out. Routing unknown actions through batched review is the safer\n",
    "    default than auto-executing them.\n",
    "    \"\"\"\n",
    "    text = action.lower()\n",
    "    if any(kw in text for kw in HIGH_RISK_KEYWORDS):\n",
    "        return \"high\"\n",
    "    if any(kw in text for kw in LOW_RISK_KEYWORDS):\n",
    "        return \"low\"\n",
    "    if any(kw in text for kw in MEDIUM_RISK_KEYWORDS):\n",
    "        return \"medium\"\n",
    "    # Explicit fail-safe default: unrecognized actions route to batched review.\n",
    "    return \"medium\"\n",
    "\n",
    "\n",
    "def tiered_gate(action: str, attempt: int = 0) -> dict:\n",
    "    \"\"\"Classify then gate. Low and medium tiers auto-approve; high blocks.\"\"\"\n",
    "    tier = classify_risk(action)\n",
    "    if tier in AUTO_APPROVE_REASONS:\n",
    "        return {\n",
    "            \"decision\": \"approve\",\n",
    "            \"reason\": AUTO_APPROVE_REASONS[tier],\n",
    "            \"action\": action,\n",
    "            \"risk_tier\": tier,\n",
    "            \"ts\": datetime.now(timezone.utc).isoformat(),\n",
    "        }\n",
    "    return gate_action(action, tier, attempt=attempt)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 模式 3：审计日志和修订循环\n",
    "\n",
    "一个 `print(\"Response approved.\")` 不是审计日志。为了建立信任，每个门禁决策都应该作为结构化事件记录，您可以稍后查询、重放或附加到事件审查中。\n",
    "\n",
    "两个部分：\n",
    "\n",
    "1. **仅追加的 JSONL。** 每个决策一行，包含时间戳、操作、层级、决策、原因。易于 grep，之后也容易发送到真正的日志存储中。\n",
    "2. **拒绝时的修订循环。** 当门禁返回 `deny` 时，代理会在上下文中带入拒绝原因重新提示自己，以便下一次提案能避免该问题。\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def log_decision(decision: dict) -> None:\n",
    "    \"\"\"Append a gate decision to the JSONL audit log.\"\"\"\n",
    "    with GATE_LOG_PATH.open(\"a\", encoding=\"utf-8\") as f:\n",
    "        f.write(json.dumps(decision) + \"\\n\")\n",
    "\n",
    "\n",
    "def propose_action(goal: str, prior_rejection: str | None = None) -> str:\n",
    "    \"\"\"Ask the LLM to propose a concrete next action for a goal.\n",
    "\n",
    "    If prior_rejection is provided, it is fed back so the LLM can avoid\n",
    "    the same failure mode in the next proposal.\n",
    "    \"\"\"\n",
    "    system = (\n",
    "        \"You are an action planner for an agent. Propose ONE concrete next\\n\"\n",
    "        \"action (a single sentence) toward the user's goal. If a prior\\n\"\n",
    "        \"rejection reason is given, propose a different action that addresses\\n\"\n",
    "        \"the rejection.\"\n",
    "    )\n",
    "    user_text = f\"Goal: {goal}\"\n",
    "    if prior_rejection:\n",
    "        user_text += f\"\\n\\nPrior proposal was denied. Reason: {prior_rejection}\"\n",
    "\n",
    "    response = client.responses.create(\n",
    "        model=deployment,\n",
    "        input=[\n",
    "            {\"role\": \"system\", \"content\": system},\n",
    "            {\"role\": \"user\", \"content\": user_text},\n",
    "        ],\n",
    "        store=False,\n",
    "    )\n",
    "    return response.output_text.strip()\n",
    "\n",
    "\n",
    "def run_with_revision(goal: str, max_revisions: int = 2) -> dict:\n",
    "    \"\"\"Propose, gate, and on rejection revise up to max_revisions times.\"\"\"\n",
    "    prior_reason: str | None = None\n",
    "    for attempt in range(max_revisions + 1):\n",
    "        action = propose_action(goal, prior_rejection=prior_reason)\n",
    "        decision = tiered_gate(action, attempt=attempt)\n",
    "        decision[\"attempt\"] = attempt\n",
    "        log_decision(decision)\n",
    "        if decision[\"decision\"] == \"approve\":\n",
    "            return decision\n",
    "        prior_reason = decision[\"reason\"]\n",
    "    return {**decision, \"final\": \"max_revisions_reached\"}\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# End-to-end demo: three goals at three different risk profiles.\n",
    "# GATE_LOG_PATH is per-run (timestamped) so no prior log is touched.\n",
    "\n",
    "goals = [\n",
    "    \"Look up the weather in Seattle for the customer's trip planning.\",\n",
    "    \"Schedule a reminder for the customer to check in 24 hours before their flight.\",\n",
    "    \"Send a marketing email to the customer about premium upgrade options.\",\n",
    "]\n",
    "\n",
    "for goal in goals:\n",
    "    print(f\"\\n=== Goal: {goal} ===\")\n",
    "    outcome = run_with_revision(goal, max_revisions=1)\n",
    "    print(f\"[final] {outcome['decision']} ({outcome['reason']})\")\n",
    "\n",
    "print(f\"\\n=== Audit log ({GATE_LOG_PATH.name}) ===\")\n",
    "for line in GATE_LOG_PATH.read_text(encoding=\"utf-8\").splitlines():\n",
    "    record = json.loads(line)\n",
    "    print(f\"  [{record['risk_tier']:6s}] {record['decision']:8s} \"\n",
    "          f\"attempt={record.get('attempt', '?')} action={record['action'][:140]}\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 额外资源\n",
    "\n",
    "其他几个公共项目实现了这些 HITL 模式的各种变体。比较不同方法，找到适合您技术栈的：\n",
    "\n",
    "- **LangChain** 人类在环工具封装 ([docs](https://python.langchain.com/docs/integrations/tools/human_tools)): 可即插即用的工具封装，暂停执行等待人类输入。\n",
    "- **AutoGen** `UserProxyAgent` ([v0.2 docs](https://microsoft.github.io/autogen/0.2/docs/topics/human-in-the-loop); AutoGen v0.4+ 重构了此部分): 使用代理角色专门代表多代理对话中的人类。\n",
    "- **Microsoft Agent Framework (MAF)** 函数调用中间件 ([docs](https://learn.microsoft.com/agent-framework/)): 在每个工具/函数调用周围运行的中间件，适合用于门控逻辑和审批流程。\n",
    "\n",
    "每个项目对三种子模式的处理方式不同：LangChain 将它们封装为工具，AutoGen 使用代理角色，Microsoft Agent Framework 则使用函数调用中间件。在选择自己代理的设计之前，请完整阅读一个或两个实现方案。\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---\n\n<!-- CO-OP TRANSLATOR DISCLAIMER START -->\n**免责声明**：\n本文件由 AI 翻译服务 [Co-op Translator](https://github.com/Azure/co-op-translator) 翻译完成。尽管我们力求准确，但请注意，自动翻译可能包含错误或不准确之处。原始语言版文件应视为权威来源。对于重要信息，建议使用专业人工翻译。我们对因使用本翻译而产生的任何误解或误释不承担责任。\n<!-- CO-OP TRANSLATOR DISCLAIMER END -->\n"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "name": "python",
   "version": "3.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}