[智能体设计模式] 第13章：人类参与环节（HITL）

HITL 模式的核心是人机协同，将 AI 的高效计算能力与人类的判断力、创造力结合，在高风险、复杂或模糊场景中确保决策质量、伦理合规和安全可控，而非完全替代人类。

注意事项

HITL 模式存在以下核心缺点：

• 可扩展性不足：人类操作员无法应对数百万级别的海量任务，纯人工监督难以适配大规模场景。
• 依赖专业能力：效果高度取决于人类操作员的专业水平，细微错误的识别、AI 指导修复等需专业人员，人工标注者也需专门培训才能保证纠正质量。
• 流程复杂性提升：实施涉及隐私问题，敏感信息必须经过严格匿名化处理后才能暴露给人类，额外增加了流程成本和复杂度。

核心逻辑

1. 关键作用

• 人类监督：监控 AI 输出，防止违规或不良结果。
• 干预纠正：AI 遇困时，人类介入补充数据、引导方向。
• 反馈学习：人类反馈用于优化模型（如 RLHF）。
• 决策把关：AI 提供建议，人类做最终关键决策。
• 升级机制：超出 AI 能力的任务自动转人工。

2. 典型应用场景

• 内容审核：AI 初筛违规内容，模糊边界案例转人工裁决。
• 金融欺诈检测：AI 标记可疑交易，高风险案例由人类分析师复核。
• 客户支持：常规问题 AI 处理，复杂/高情绪问题转人工客服。
• 法律文档审查：AI 扫描分类文档，人类律师确认关键条款和法律影响。
• 自动驾驶：常规路况 AI 自主驾驶，复杂场景移交人类驾驶员。

实战代码示例（LangChain）

以下基于 LangChain 实现技术支持智能体，核心能力：处理常规技术问题，遇到复杂场景自动触发人工升级流程，同时支持客户个性化响应。

依赖安装

pip install langchain langchain-google-genai python-dotenv

创建 .env 文件配置密钥：

GOOGLE_API_KEY=你的 Gemini API 密钥（从 Google AI Studio 获取）

核心代码

"""
LangChain HITL 模式示例：技术支持智能体
核心特性：
1. 常规技术问题：AI 自主排查、创建工单，高效处理
2. 复杂问题：自动触发人工升级流程，确保问题解决质量
3. 个性化响应：结合客户信息（姓名、等级、购买记录）定制回复
"""import logging
from typing import Dict, Optional
from dotenv import load_dotenv
from langchain.agents import AgentType, initialize_agent, Tool
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain.schema import HumanMessage, SystemMessage
from langchain.callbacks.base import BaseCallbackHandler# --------------------------
# 1. 日志配置：记录流程与人工升级事件
# --------------------------
logging.basicConfig(level=logging.INFO,format="%(asctime)s - %(levelname)s - %(message)s",handlers=[logging.FileHandler("hitl_support.log"), logging.StreamHandler()]
)
logger = logging.getLogger("support_agent")# 加载环境变量
load_dotenv()# --------------------------
# 2. 模拟技术支持工具（实际场景对接真实系统）
# --------------------------
def troubleshoot_issue(issue: str) -> str:"""工具：基础故障排查功能：为常见技术问题提供标准化排查步骤适用场景：简单、可标准化的技术问题（如网络连接、软件启动失败）"""logger.info(f"执行基础故障排查，问题：{issue}")# 模拟常见问题的排查逻辑common_issues = {"网络连接": "1. 检查路由器电源是否插好；2. 重启设备；3. 忘记网络后重新连接","软件崩溃": "1. 关闭后台多余程序；2. 重启软件；3. 检查软件更新","无法开机": "1. 检查电源适配器是否通电；2. 长按电源键10秒强制重启；3. 联系售后检测硬件"}# 匹配常见问题，无匹配则返回通用排查步骤steps = common_issues.get(issue[:5], "1. 重启相关设备；2. 检查网络连接；3. 确认软件版本最新")return f"基础故障排查步骤：\n{steps}\n\n若仍未解决，请等待人工专家跟进。"def create_ticket(issue_type: str, details: str) -> str:"""工具：创建服务工单功能：记录未解决的问题，生成唯一工单ID，便于后续跟踪"""import randomticket_id = f"TICKET-{random.randint(1000, 9999)}"logger.info(f"创建服务工单，类型：{issue_type}，详情：{details}，工单ID：{ticket_id}")return f"服务工单已创建！\n工单ID：{ticket_id}\n问题类型：{issue_type}\n问题详情：{details}\n我们将尽快处理。"def escalate_to_human(issue_type: str, customer_info: str) -> str:"""工具：人工升级功能：将复杂问题转至人工客服队列，返回升级结果提示触发场景：AI 无法处理的复杂问题、客户明确要求人工、高等级客户问题"""logger.warning(f"触发人工升级！问题类型：{issue_type}，客户信息：{customer_info}")return (f"您好！您的问题（{issue_type}）已升级至人工专家处理。\n"f"基于您的客户信息：{customer_info}\n"f"预计响应时间：15分钟内（工作时间9:00-18:00）\n"f"如需紧急处理，可拨打人工热线：400-XXX-XXXX")# --------------------------
# 3. 回调处理器：客户信息个性化注入
# --------------------------
class PersonalizationCallback(BaseCallbackHandler):"""自定义回调处理器：核心作用：在智能体生成响应前，注入客户个性化信息，提升回复针对性"""def __init__(self, customer_info: Dict[str, str]):# 客户信息字典（实际场景从数据库/接口获取）self.customer_info = customer_info or {"name": "尊贵客户","tier": "标准会员","recent_purchases": "无"}def on_chain_start(self, serialized: Dict[str, Any], inputs: Dict[str, Any], **kwargs):"""智能体开始执行时触发，注入个性化系统消息"""logger.info("注入客户个性化信息")# 构造个性化提示personalization_note = (f"客户姓名：{self.customer_info['name']}\n"f"客户等级：{self.customer_info['tier']}\n"f"最近购买：{self.customer_info['recent_purchases']}\n"f"回复要求：称呼客户姓名，高等级客户（VIP/钻石）优先承诺更快响应")# 将个性化信息添加到输入的系统消息中if "messages" in inputs:inputs["messages"].insert(0, SystemMessage(content=personalization_note))# --------------------------
# 4. 构建 HITL 技术支持智能体
# --------------------------
def build_support_agent(customer_info: Optional[Dict[str, str]] = None) -> Any:"""构建技术支持智能体参数：customer_info - 客户信息字典（姓名、等级、最近购买等）返回：初始化后的 LangChain 智能体"""# 初始化 LLM 模型llm = ChatGoogleGenerativeAI(model="gemini-2.0-flash",temperature=0.3  # 低温度保证响应一致性和专业性)# 初始化个性化回调personalization_callback = PersonalizationCallback(customer_info or {})# 定义工具集tools = [Tool(name="troubleshoot_issue",func=troubleshoot_issue,description="用于处理简单、常见的技术问题（如网络连接、软件崩溃、无法开机），提供标准化排查步骤"),Tool(name="create_ticket",func=create_ticket,description="用于记录未解决的问题，生成服务工单，需传入问题类型和详细描述"),Tool(name="escalate_to_human",func=lambda issue_type: escalate_to_human(issue_type, str(customer_info)),description="用于处理复杂、AI 无法解决的问题，或客户明确要求人工服务时，需传入问题类型")]# 初始化智能体agent = initialize_agent(tools=tools,llm=llm,agent=AgentType.CHAT_ZERO_SHOT_REACT_DESCRIPTION,verbose=True,callbacks=[personalization_callback],agent_kwargs={"system_message": """你是专业的电子产品技术支持专家，工作流程如下：1. 先理解客户的技术问题，判断问题复杂度：- 简单问题（常见、可标准化）：调用 troubleshoot_issue 工具提供排查步骤- 复杂问题（无标准化解决方案、涉及硬件故障、客户多次反馈未解决）：直接调用 escalate_to_human 工具升级人工2. 若客户按排查步骤操作后仍未解决，调用 create_ticket 工具创建工单，并告知客户工单ID3. 回复语气：专业、有同理心，认可客户的困扰，步骤清晰易懂4. 优先使用客户提供的个性化信息（姓名、等级）定制回复5. 客户明确要求"转人工"时，立即调用 escalate_to_human 工具，无需额外处理"""})return agent# --------------------------
# 5. 测试运行：验证不同场景下的 HITL 逻辑
# --------------------------
if __name__ == "__main__":print("=== HITL 技术支持智能体测试 ===\n")# 测试用例 1：简单技术问题（AI 自主处理）print("【测试用例 1】简单技术问题")customer1 = {"name": "张三","tier": "标准会员","recent_purchases": "XX品牌笔记本电脑（3个月前）"}agent1 = build_support_agent(customer1)query1 = "我的笔记本电脑网络连接不上，怎么办？"print(f"客户输入：{query1}")print(f"智能体响应：{agent1.run(query1)}\n")# 测试用例 2：复杂问题（触发人工升级）print("【测试用例 2】复杂技术问题")customer2 = {"name": "李四","tier": "VIP会员","recent_purchases": "XX品牌台式机（1个月前）"}agent2 = build_support_agent(customer2)query2 = "电脑开机后蓝屏，显示未知错误代码0x0000007B，重启多次仍无效"print(f"客户输入：{query2}")print(f"智能体响应：{agent2.run(query2)}\n")# 测试用例 3：客户明确要求转人工print("【测试用例 3】客户要求转人工")customer3 = {"name": "王五","tier": "钻石会员","recent_purchases": "XX品牌平板电脑（1周前）"}agent3 = build_support_agent(customer3)query3 = "我想直接找人工客服处理，我的平板触摸失灵了"print(f"客户输入：{query3}")print(f"智能体响应：{agent3.run(query3)}")

核心亮点

1. 分层处理：简单问题 AI 自主解决，复杂问题自动转人工，兼顾效率与质量。
2. 个性化体验：通过回调处理器注入客户信息，回复更具针对性（如称呼姓名、VIP 优先响应）。
3. 可追溯性：完整日志记录问题处理流程和人工升级事件，便于后续复盘。
4. 灵活扩展：工具与智能体逻辑分离，可轻松新增故障类型、升级规则或对接真实业务系统。