AI Agent 人机协作模式深度解析：Human-in-the-Loop 架构设计与实践

当 AI Agent 越来越强大，人类该如何保持主导权？Human-in-the-Loop（人机协作）不是阻碍效率的枷锁，而是构建可信 AI 系统的关键架构模式。本文从技术实现到产品设计，深度剖析人机协作的核心机制与最佳实践。

引言：为什么需要人机协作？

2025 年，AI Agent 的能力边界正在快速扩展。从代码生成到复杂决策，从数据分析到内容创作，Agent 正在接管越来越多的认知任务。但一个核心问题始终悬而未决：当 Agent 犯错时，谁来把关？

纯粹自主的 Agent 面临三重困境：

1. 可靠性边界：大语言模型的幻觉问题尚未根治，关键决策场景下完全放手风险极高
2. 价值对齐：Agent 的优化目标可能与人类真实意图存在偏差，需要持续校准
3. 责任归属：当 Agent 造成损失时，责任如何界定？人类的参与是信任的基础

Human-in-the-Loop（HITL，人机协作）架构正是在这种背景下成为必选项。它不是对 Agent 能力的否定，而是对可控性的强化。本文将从技术架构、实现模式、最佳实践三个维度，系统解析如何构建高效的人机协作系统。

一、人机协作的核心概念与模式

1.1 人机协作的本质

人机协作不是简单的人工审核，而是一种状态感知的交互架构。它的核心特征包括：

特征	描述
主动性	系统主动识别需要人类介入的场景，而非被动等待
上下文保留	人类介入时能看到完整的决策背景和推理过程
双向反馈	人类输入不仅修正当前决策，还能优化 Agent 的未来行为
渐进式放权	随着信任建立，逐步减少人工介入的频率和深度

1.2 三种协作模式

根据人类介入的深度和时机，人机协作可分为三种典型模式：

模式 A：审批模式 (Approval Mode)

场景：高风险操作执行前必须获得人工确认

典型应用：

• 金融转账超过阈值
• 生产环境配置变更
• 客户敏感信息修改

核心机制：

# 伪代码示意
if operation.risk_level > RISK_THRESHOLD:
    human_decision = await request_approval(context)
    if not human_decision.approved:
        return OperationResult.rejected(human_decision.reason)

模式 B：编辑模式 (Edit Mode)

场景：Agent 生成内容，人类进行修改完善

典型应用：

• 营销文案生成与润色
• 代码生成与审查
• 数据分析报告编写

核心机制：

draft = agent.generate_content(request)
reviewed = await human_review_and_edit(draft)
final = agent.refine(reviewed.feedback)

模式 C：协作模式 (Collaborative Mode)

场景：人类与 Agent 实时协作完成任务

典型应用：

• 交互式数据分析
• 复杂问题逐步求解
• 创意头脑风暴

核心机制：

while task.not_complete():
    agent_suggestion = agent.propose_next_step(state)
    human_input = await get_human_input(agent_suggestion)
    state = state.evolve(human_input)

二、LangGraph 中的 HITL 实现

LangGraph 作为目前最成熟的 Agent 编排框架，提供了完整的 HITL 支持。其设计哲学是：中断与恢复是工作流的一等公民。

2.1 Interrupt 机制详解

LangGraph 的 interrupt 函数是 HITL 的核心原语：

from langgraph.types import interrupt

def human_approval_node(state: State):
    """需要人工审批的节点"""
    operation = state["pending_operation"]
    
    # 中断执行，等待人类输入
    human_response = interrupt({
        "question": f"批准以下操作？{operation.description}",
        "operation": operation.to_dict(),
        "context": state["context"]
    })
    
    # 收到人类响应后继续
    if human_response["decision"] == "approve":
        state["approved"] = True
        state["execution_result"] = execute_operation(operation)
    else:
        state["approved"] = False
        state["rejection_reason"] = human_response["reason"]
    
    return state

关键特性：

1. 状态持久化：中断时，完整的状态会被持久化存储，系统可以无感知恢复
2. 多模态支持：interrupt 可以传递任意可序列化的数据，包括文本、图片、结构化数据
3. 并行处理：多个中断可以并行等待，适合多人协作场景

2.2 工作流集成

将 HITL 节点集成到 LangGraph 工作流中：

from langgraph.graph import StateGraph, START, END
from typing import TypedDict, Annotated
import operator

class State(TypedDict):
    messages: Annotated[list, operator.add]
    pending_tool_calls: list
    approved_tools: list
    rejected_tools: list

def agent_decision_node(state: State):
    """Agent 决策节点"""
    response = llm.invoke(state["messages"])
    
    # 检查是否有需要审批的工具调用
    tool_calls = extract_tool_calls(response)
    high_risk_tools = filter_high_risk(tool_calls)
    
    if high_risk_tools:
        return {
            "messages": [response],
            "pending_tool_calls": high_risk_tools
        }
    
    # 低风险工具直接执行
    results = execute_safe_tools(tool_calls)
    return {
        "messages": [response, results],
        "pending_tool_calls": []
    }

def human_review_node(state: State):
    """人工审核节点"""
    pending = state["pending_tool_calls"]
    
    approved = []
    rejected = []
    
    for tool in pending:
        decision = interrupt({
            "type": "tool_approval",
            "tool": tool,
            "arguments": tool["arguments"],
            "risk_score": calculate_risk(tool)
        })
        
        if decision["approved"]:
            approved.append(tool)
        else:
            rejected.append({
                "tool": tool,
                "reason": decision["reason"]
            })
    
    return {
        "approved_tools": approved,
        "rejected_tools": rejected
    }

def execute_approved_node(state: State):
    """执行已批准的工具"""
    results = []
    for tool in state["approved_tools"]:
        result = execute_tool(tool)
        results.append(result)
    
    return {"messages": results}

# 构建工作流
builder = StateGraph(State)
builder.add_node("agent", agent_decision_node)
builder.add_node("human_review", human_review_node)
builder.add_node("execute", execute_approved_node)

builder.add_edge(START, "agent")
builder.add_conditional_edges(
    "agent",
    lambda s: "human_review" if s["pending_tool_calls"] else END,
    {"human_review": "human_review", END: END}
)
builder.add_edge("human_review", "execute")
builder.add_edge("execute", END)

graph = builder.compile(checkpointer=checkpointer)

2.3 中断恢复机制

LangGraph 的中断恢复依赖于检查点 (Checkpointer) 机制：

from langgraph.checkpoint.memory import MemorySaver

# 配置检查点
memory = MemorySaver()
graph = builder.compile(checkpointer=memory)

# 启动工作流（首次执行）
thread_id = "unique-thread-id"
config = {"configurable": {"thread_id": thread_id}}

for event in graph.stream({"messages": [user_input]}, config):
    print(event)
    # 执行到 interrupt 时会暂停，等待人类输入

# ====== 中断后恢复 ======
# 人类通过某种方式提供了响应
human_response = {"decision": "approve", "reason": ""}

# 使用 Command 恢复执行
from langgraph.types import Command

for event in graph.stream(
    Command(resume=human_response),
    config
):
    print(event)

恢复机制的关键设计：

1. 确定性重放：从检查点恢复后，已执行的节点不会重复执行
2. 幂等性保证：即使网络故障导致重复恢复，系统状态保持一致
3. 历史可见：人类介入时可以看到完整的执行历史和上下文

三、HITL 的产品化设计

技术实现只是基础，真正的人机协作需要在产品层面精心设计。

3.1 智能中断触发

不是所有决策都需要人工介入，关键在于精准识别高风险时刻：

class RiskAssessor:
    """风险评估器"""
    
    def __init__(self):
        self.rules = [
            # 基于操作类型的规则
            OperationTypeRule(),
            # 基于参数敏感度的规则
            ParameterSensitivityRule(),
            # 基于历史错误率的规则
            HistoricalErrorRule(),
            # 基于用户信任度的规则
            UserTrustRule()
        ]
    
    def assess(self, operation: Operation, context: Context) -> RiskScore:
        scores = [rule.evaluate(operation, context) for rule in self.rules]
        return RiskScore.combine(scores)
    
    def should_interrupt(self, operation: Operation, context: Context) -> bool:
        score = self.assess(operation, context)
        # 动态阈值：根据场景和用户调整
        threshold = self.get_adaptive_threshold(context)
        return score.value > threshold

触发策略矩阵：

维度	低风险（自动）	中风险（提示但不阻断）	高风险（必须审批）
操作类型	查询类	创建类	删除/修改关键数据
影响范围	个人数据	团队数据	全平台数据
不可逆性	可撤销	撤销成本高	不可撤销
合规要求	无	需审计日志	需双人复核

3.2 上下文呈现设计

人类介入时，需要看到足够但不过载的信息：

def build_review_context(state: State, operation: Operation) -> ReviewContext:
    """构建审核上下文"""
    
    context = ReviewContext()
    
    # 核心信息：决策是什么
    context.decision_summary = operation.summary()
    
    # 推理过程：Agent 为什么这样决策
    context.reasoning_trace = extract_reasoning(state["messages"])
    
    # 影响评估：如果执行会发生什么
    context.impact_preview = simulate_impact(operation)
    
    # 替代方案：还有其他选择吗
    context.alternatives = generate_alternatives(operation)
    
    # 历史参考：类似情况过去怎么处理
    context.similar_cases = query_historical_decisions(operation)
    
    return context

界面设计原则：

1. 分层展示：默认展示核心信息，详细信息可展开
2. 视觉编码：用颜色和图标快速传达风险等级
3. 对比视图：将 Agent 建议与替代方案并排展示
4. 渐进式披露：复杂信息按需展示，避免认知过载

3.3 反馈学习闭环

人类介入不仅是修正当前决策，更是训练 Agent 的重要机会：

class FeedbackLearner:
    """从人类反馈中学习"""
    
    def capture_feedback(
        self,
        original_decision: Decision,
        human_correction: Correction,
        context: Context
    ):
        """捕获人类修正"""
        
        feedback = {
            "timestamp": now(),
            "context": context,
            "original": original_decision,
            "correction": human_correction,
            "deviation_type": self.classify_deviation(original_decision, human_correction)
        }
        
        # 存储反馈
        self.feedback_store.save(feedback)
        
        # 触发模型微调或提示词优化
        if self.should_trigger_training(feedback):
            self.schedule_fine_tuning()
    
    def update_risk_model(self):
        """基于反馈更新风险评估模型"""
        recent_feedback = self.feedback_store.get_recent(days=30)
        
        # 识别误判模式
        false_positives = [f for f in recent_feedback 
                          if f.deviation_type == "unnecessary_interrupt"]
        false_negatives = [f for f in recent_feedback 
                          if f.deviation_type == "missed_risk"]
        
        # 调整风险阈值
        self.risk_assessor.adjust_thresholds(false_positives, false_negatives)

四、实战：构建一个智能客服 Agent 的人机协作系统

让我们通过一个完整的案例，展示如何在实际项目中落地 HITL 架构。

4.1 场景分析

业务场景：智能客服 Agent 处理用户退款请求

协作需求：

• 小额退款 (< $50)：自动处理
• 中等金额 ($50-$500)：提示客服，但允许批量审批
• 大额退款 (≥ $500)：必须人工逐单审批
• 异常模式：用户短期内多次退款，触发人工审查

4.2 完整代码实现

from dataclasses import dataclass
from enum import Enum
from typing import Optional
from langgraph.graph import StateGraph, START, END
from langgraph.types import interrupt, Command
from langgraph.checkpoint.memory import MemorySaver

class RefundTier(Enum):
    AUTO = "auto"          # 自动处理
    NOTIFY = "notify"      # 通知但不阻断
    APPROVE = "approve"    # 必须审批
    REVIEW = "review"      # 异常审查

@dataclass
class RefundRequest:
    order_id: str
    user_id: str
    amount: float
    reason: str
    request_time: datetime
    user_history: UserHistory

@dataclass
class UserHistory:
    total_orders: int
    total_refunds: int
    refund_rate: float
    recent_refunds: list[RefundRecord]

class State(TypedDict):
    request: RefundRequest
    tier: RefundTier
    decision: Optional[RefundDecision]
    human_feedback: Optional[dict]

class RiskEngine:
    """退款风险评估引擎"""
    
    def assess(self, request: RefundRequest) -> RefundTier:
        # 规则 1：金额分级
        if request.amount >= 500:
            return RefundTier.APPROVE
        elif request.amount >= 50:
            base_tier = RefundTier.NOTIFY
        else:
            base_tier = RefundTier.AUTO
        
        # 规则 2：用户历史检查
        history = request.user_history
        
        # 退款率过高
        if history.refund_rate > 0.3:
            return RefundTier.REVIEW
        
        # 短期内多次退款
        if len(history.recent_refunds) > 2:
            return RefundTier.REVIEW
        
        # 新用户大额退款
        if history.total_orders < 3 and request.amount > 100:
            return RefundTier.APPROVE
        
        return base_tier

# ========== 工作流节点定义 ==========

def assess_risk_node(state: State):
    """风险评估节点"""
    engine = RiskEngine()
    tier = engine.assess(state["request"])
    return {"tier": tier}

def auto_process_node(state: State):
    """自动处理节点"""
    request = state["request"]
    
    # 执行退款
    result = execute_refund(request)
    
    return {
        "decision": RefundDecision(
            approved=True,
            auto=True,
            executed_at=now(),
            result=result
        )
    }

def notify_only_node(state: State):
    """仅通知节点"""
    request = state["request"]
    
    # 发送通知但不阻断
    send_notification(
        channel="slack",
        message=f"退款已自动处理: 订单 {request.order_id}, 金额 ${request.amount}"
    )
    
    # 继续自动处理
    result = execute_refund(request)
    
    return {
        "decision": RefundDecision(
            approved=True,
            auto=True,
            notified=True,
            result=result
        )
    }

def human_approval_node(state: State):
    """人工审批节点"""
    request = state["request"]
    
    # 构建审批上下文
    context = {
        "订单信息": {
            "订单号": request.order_id,
            "用户ID": request.user_id,
            "退款金额": f"${request.amount}",
            "退款原因": request.reason
        },
        "用户历史": {
            "总订单数": request.user_history.total_orders,
            "历史退款数": request.user_history.total_refunds,
            "退款率": f"{request.user_history.refund_rate:.1%}",
            "近期退款": [
                f"{r.date}: ${r.amount}" 
                for r in request.user_history.recent_refunds
            ]
        },
        "风险评估": f"金额 ${request.amount} 超过自动处理阈值"
    }
    
    # 中断等待人工输入
    human_input = interrupt({
        "type": "refund_approval",
        "title": f"退款审批: 订单 {request.order_id}",
        "context": context,
        "actions": ["approve", "reject", "escalate"]
    })
    
    # 处理人类反馈
    if human_input["action"] == "approve":
        result = execute_refund(request)
        decision = RefundDecision(
            approved=True,
            auto=False,
            approved_by=human_input["user_id"],
            result=result
        )
    elif human_input["action"] == "reject":
        decision = RefundDecision(
            approved=False,
            auto=False,
            rejected_by=human_input["user_id"],
            reason=human_input["reason"]
        )
    else:  # escalate
        decision = RefundDecision(
            approved=False,
            escalated=True,
            escalated_to="senior_support"
        )
    
    return {"decision": decision, "human_feedback": human_input}

def fraud_review_node(state: State):
    """异常审查节点"""
    request = state["request"]
    
    # 构建更详细的审查信息
    context = {
        "alert_type": "potential_fraud",
        "risk_factors": [
            f"退款率异常: {request.user_history.refund_rate:.1%}",
            f"短期多次退款: {len(request.user_history.recent_refunds)} 次"
        ],
        "recommended_action": "人工审查后再决定"
    }
    
    human_input = interrupt({
        "type": "fraud_review",
        "title": f"异常退款审查: 用户 {request.user_id}",
        "context": context,
        "actions": ["approve", "reject", "investigate"]
    })
    
    # 处理逻辑...
    return {"decision": decision, "human_feedback": human_input}

def finalize_node(state: State):
    """完结节点"""
    decision = state["decision"]
    
    # 记录审计日志
    log_audit_event({
        "order_id": state["request"].order_id,
        "decision": decision.to_dict(),
        "human_feedback": state.get("human_feedback")
    })
    
    # 通知用户
    notify_user(state["request"].user_id, decision)
    
    return {"status": "completed"}

# ========== 构建工作流 ==========

def route_by_tier(state: State):
    """根据风险等级路由"""
    tier = state["tier"]
    
    routing = {
        RefundTier.AUTO: "auto_process",
        RefundTier.NOTIFY: "notify_only",
        RefundTier.APPROVE: "human_approval",
        RefundTier.REVIEW: "fraud_review"
    }
    
    return routing[tier]

# 构建图
builder = StateGraph(State)

builder.add_node("assess_risk", assess_risk_node)
builder.add_node("auto_process", auto_process_node)
builder.add_node("notify_only", notify_only_node)
builder.add_node("human_approval", human_approval_node)
builder.add_node("fraud_review", fraud_review_node)
builder.add_node("finalize", finalize_node)

builder.add_edge(START, "assess_risk")
builder.add_conditional_edges(
    "assess_risk",
    route_by_tier,
    {
        "auto_process": "auto_process",
        "notify_only": "notify_only",
        "human_approval": "human_approval",
        "fraud_review": "fraud_review"
    }
)
builder.add_edge("auto_process", "finalize")
builder.add_edge("notify_only", "finalize")
builder.add_edge("human_approval", "finalize")
builder.add_edge("fraud_review", "finalize")
builder.add_edge("finalize", END)

# 编译
memory = MemorySaver()
refund_graph = builder.compile(checkpointer=memory)

4.3 前端集成示例

// React 组件：退款审批界面
interface RefundApprovalProps {
  threadId: string;
  onComplete: () => void;
}

const RefundApproval: React.FC<RefundApprovalProps> = ({ threadId, onComplete }) => {
  const [interruptData, setInterruptData] = useState<InterruptData | null>(null);
  
  useEffect(() => {
    // 监听中断事件
    const eventSource = new EventSource(
      `/api/graph/events?thread_id=${threadId}`
    );
    
    eventSource.onmessage = (event) => {
      const data = JSON.parse(event.data);
      if (data.type === 'interrupt') {
        setInterruptData(data.payload);
      }
      if (data.type === 'end') {
        onComplete();
      }
    };
    
    return () => eventSource.close();
  }, [threadId]);
  
  const handleDecision = async (action: string, reason?: string) => {
    await fetch('/api/graph/resume', {
      method: 'POST',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify({
        thread_id: threadId,
        resume_data: { action, reason, user_id: currentUser.id }
      })
    });
    setInterruptData(null);
  };
  
  if (!interruptData) return null;
  
  return (
    <div className="approval-modal">
      <h2>{interruptData.title}</h2>
      
      <div className="context-section">
        <h3>订单信息</h3>
        <dl>
          {Object.entries(interruptData.context.订单信息).map(([k, v]) => (
            <div key={k}>
              <dt>{k}</dt>
              <dd>{v}</dd>
            </div>
          ))}
        </dl>
      </div>
      
      <div className="risk-warning">
        <AlertTriangle />
        {interruptData.context.风险评估}
      </div>
      
      <div className="actions">
        {interruptData.actions.map(action => (
          <button
            key={action}
            onClick={() => handleDecision(action)}
            className={`btn-${action}`}
          >
            {action === 'approve' && '✓ 批准'}
            {action === 'reject' && '✗ 拒绝'}
            {action === 'escalate' && '↑ 升级'}
          </button>
        ))}
      </div>
    </div>
  );
};

五、人机协作的进阶模式

5.1 多智能体 + 多人协作

复杂场景下，可能需要多个 Agent 与多个人类协作：

from langgraph.graph import StateGraph
from typing import Literal

class MultiAgentState(TypedDict):
    case_id: str
    findings: dict
    human_inputs: dict[str, dict]  # role -> input
    consensus_reached: bool

def specialist_agent(state: State, domain: str):
    """领域专家 Agent"""
    analysis = llm.invoke(
        f"作为{domain}专家，分析以下情况：{state['findings']}"
    )
    return {"findings": {domain: analysis}}

def coordinator_node(state: State):
    """协调节点：整合多个专家意见"""
    
    # 检查是否需要人类专家介入
    if has_conflict(state["findings"]):
        # 并行中断，等待多个人类专家
        inputs = interrupt({
            "type": "multi_expert_review",
            "findings": state["findings"],
            "required_roles": ["legal", "technical", "business"]
        })
        
        state["human_inputs"] = inputs
        state["consensus_reached"] = evaluate_consensus(inputs)
    
    return state

5.2 渐进式放权

随着系统运行，可以逐步减少人工介入：

class AdaptiveHITL:
    """自适应人机协作"""
    
    def __init__(self):
        self.accuracy_history = deque(maxlen=100)
        self.current_threshold = 0.8
    
    def should_interrupt(self, confidence: float, impact: float) -> bool:
        """动态决定是否中断"""
        
        # 基础规则
        if impact > CRITICAL_IMPACT:
            return True
        
        # 根据历史准确率调整阈值
        if self.accuracy_history:
            avg_accuracy = sum(self.accuracy_history) / len(self.accuracy_history)
            # 准确率高时可以更激进
            if avg_accuracy > 0.95:
                effective_threshold = self.current_threshold * 0.9
            else:
                effective_threshold = self.current_threshold
        else:
            effective_threshold = self.current_threshold
        
        return confidence < effective_threshold
    
    def record_outcome(self, prediction_correct: bool):
        """记录预测结果"""
        self.accuracy_history.append(prediction_correct)

5.3 时间感知的中断

某些场景下，中断需要考虑时间因素：

def time_aware_interrupt(state: State):
    """时间感知的中断"""
    
    request_time = state["request_time"]
    current_time = now()
    elapsed = current_time - request_time
    
    # 如果已经等待太久，自动降级处理
    if elapsed > MAX_WAIT_TIME:
        return auto_approve_with_escalation(state)
    
    # 工作时间 vs 非工作时间不同策略
    if is_business_hours():
        # 正常工作时间内，严格审批
        return strict_interrupt(state)
    else:
        # 非工作时间，降低阈值或自动处理低风险项
        if state["risk_score"] < NIGHT_MODE_THRESHOLD:
            return auto_approve(state)
        return interrupt({"urgent": True, **state})

六、最佳实践与陷阱规避

6.1 设计原则

1. 最小阻断原则

   • 只在真正必要时阻断
   • 提供批量处理机制
   • 允许规则预配置减少重复审批

2. 上下文完整性

   • 人类决策时必须有完整背景
   • 保留完整的决策历史
   • 提供”为什么”的解释

3. 渐进式信任

   • 初期高频介入建立信心
   • 根据准确率逐步放权
   • 保留关键决策的最终控制权

4. 双向学习

   • 从人类修正中提取模式
   • 定期更新风险评估模型
   • A/B 测试不同阈值策略

6.2 常见陷阱

陷阱	表现	解决方案
审批疲劳	人工介入过多导致效率下降	动态阈值 + 批量审批 + 智能预分类
上下文缺失	人类看不到完整决策背景	强制上下文构建 + 可视化推理链
反馈循环缺失	人类修正不被系统学习	建立反馈捕获和模型更新机制
单点故障	人工审批者不可用时系统卡住	多级授权 + 超时自动升级
一致性差	不同人审批标准不一	标准操作程序 + 校准会议 + 数据监控

6.3 度量指标

建立 HITL 系统的度量体系：

class HITLMetrics:
    """人机协作度量"""
    
    def collect(self):
        return {
            # 效率指标
            "avg_decision_time": measure_avg_time(),
            "interruption_rate": interrupt_count / total_requests,
            "false_positive_rate": unnecessary_interrupts / total_interrupts,
            
            # 质量指标
            "human_override_rate": overrides / auto_decisions,
            "override_accuracy": accuracy_of_overrides(),
            
            # 学习指标
            "feedback_utilization": feedback_used / feedback_collected,
            "model_improvement": accuracy_trend(),
            
            # 满意度指标
            "approver_satisfaction": survey_results(),
            "user_experience_score": ux_metrics()
        }

七、未来展望

人机协作架构正在快速演进，几个值得关注的方向：

1. 多模态中断：不仅支持文本，还支持语音、图像、视频等富媒体交互
2. 预测性介入：在 Agent 犯错之前就预判到风险并介入
3. 群体智慧：多个 Agent 和人类的协作网络，而非简单的 1:1 协作
4. 情感感知：理解人类的情绪状态，调整交互方式
5. 可解释性增强：让 Agent 的推理过程对人类更透明

结语

Human-in-the-Loop 不是权宜之计，而是生产级 AI 系统的必备架构。它体现了对 AI 能力的清醒认知：强大但非全知，高效但需要监督。

构建优秀的人机协作系统，技术实现只是 30%，剩下的 70% 在于：

• 精准的时机选择
• 完整的上下文呈现
• 流畅的交互体验
• 持续的反馈学习

最终目标是实现人机共生：Agent 负责它可以做好的，人类专注于只有人类能做好的——判断力、创造力和伦理抉择。

---

参考链接：

• LangGraph Human-in-the-Loop 文档
• Human Loop 最佳实践
• AI 安全与人机协作

---

本文首发于 Channing’s Blog，转载请注明出处。